草庐IT

data_stream

全部标签

java - 合并多个相同的 Kafka Streams 主题

我有2个Kafka主题流式传输来自不同来源的完全相同的内容,因此我可以在其中一个来源出现故障时保持高可用性。我正在尝试使用KafkaStreams0.10.1.0将2个主题合并为1个输出主题,这样我就不会错过任何有关失败的消息,并且在所有源都启动时不会出现重复。当使用KStream的leftJoin方法时,其中一个主题可以正常下降(次要主题),但是当主要主题下降时,不会向输出主题发送任何内容。这似乎是因为,根据KafkaStreamsdeveloperguide,KStream-KStreamleftJoinisalwaysdrivenbyrecordsarrivingfromthep

java - Java Stream API 是如何选择执行计划的?

我刚开始学习Java8中的StreamAPI和一般的函数式编程,但对Java并不陌生。我有兴趣了解和了解StreamAPI如何选择执行计划。它如何知道哪些部分需要并行化,哪些部分不需要?存在多少种执行计划?基本上,我想知道为什么Java8中的Streams有助于使事情变得更快,以及它如何发挥这种“魔力”。我找不到太多关于这一切如何运作的文献。 最佳答案 这个问题有点宽泛,不好详细解释,但我会尽力回答到满意的程度。我还使用了ArrayList的Stream示例。当我们创建流时,返回的对象称为ReferencePipeline.这个对象

java - Stream collect with Generic 类型

我尝试在将json反序列化为pojo的方法中使用泛型,以便它可以返回任何对象类型。这是我的代码:privateBla(Listas,Listbs){this.as=as;this.bs=bs;}publicstaticBlafrom(JsonObjectjson){returnnewBla(Bla.load(json,As),Bla.load(json,Bs));}privatestaticListload(JsonObjectjsonObject,Stringparam){returnjsonObject.getJsonArray(param).stream().map(Bla::g

java - 如何用 Streams 替换 Iterables.filter()?

我正在尝试从Guava迁移到Java8Streams,但不知道如何处理可迭代对象。这是我的代码,用于从可迭代对象中删除空字符串:Iterablelist=Iterables.filter(raw,//it'sIterablenewPredicate(){@Overridepublicbooleanapply(Stringtext){return!text.isEmpty();}});注意,这是一个Iterable,不是Collection.它可能包含无限数量的项目,我无法将它们全部加载到内存中。我的Java8替代品是什么?顺便说一句,有了Lamba,这段代码看起来会更短:Iterabl

java - 为什么Stream <T> collect方法返回不同的键顺序?

我有以下代码:publicenumContinent{ASIA,EUROPE}publicclassCountry{privateStringname;privateContinentregion;publicCountry(Stringna,Continentreg){this.name=na;this.region=reg;}publicStringgetName(){returnname;}publicContinentgetRegion(){returnregion;}@OverridepublicStringtoString(){return"Country[name="+n

java - 错误 : detached entity passed to persist - try to persist complex data (Play-Framework)

我在通过PlayFramework持久化数据时遇到问题。也许不可能达到那个结果,但如果它能奏效那就太好了。简单:我有一个复杂的模型(带地址的商店),我想立即更改带地址的商店并以相同的方式存储它们(shop.save())。但是错误detachedentitypassedtopersist发生了。更新历史05.1105.11使用属性mappedBy="shop"更新模型商店更新指向google用户组的链接09.11找到解决方法,但不是通用的16.11更新示例html表单,感谢@Pavel将解决方法(更新09.11)更新为通用方法,感谢@mericano121.11我放弃了寻找解决方案并等

java - JSch 0.1.53 session.connect() 抛出 "End of IO Stream Read"

我下载了一个新的JSch0.1.53libraryJSch(sftp)下载任务不再有效。此版本在session.connect()函数上失败并抛出错误Session.connect:java.io.IOException:EndofIOStreamRead。我的旧jsch.jar(2011-10-06)在同一台主机上工作正常,也许我缺少新的配置Prop?Sessionsession=null;ChannelSftpchannel=null;try{JSch.setLogger(SSHUtil.createJschLogger());JSchjsch=newJSch();session=

java - 为什么 Java Stream 生成器是无序的?

我尝试使用JavaStreams并行化一些工作。让我们考虑这个简单的例子:Stream.generate(newSupplier(){@OverridepublicIntegerget(){returngenerateNewInteger();}}).parallel().forEachOrdered(newConsumer(){@Overridepublicvoidaccept(Integerinteger){System.out.println(integer);}});问题是它不会为forEachOrdered调用accept方法,它只有在我使用forEach时才有效。我想问题是

PKCS11 Interop CKR_ENCRYPTED_DATA_INVALID异常原因高CPU用法

我们正在使用Pkcs11Interop密码学库,我们自定义“实施”Pkcs11Interop我们项目的代码样本Pkcs11Interop文档github。在生产中,不是每天,但是在连续使用CPU使用后,我们必须重新启动服务器(4或5次)CKR_ENCRYPTED_DATA_INVALID例外。这是什么原因?我们如何解决此问题?任何想法都将不胜感激。如果有人愿意,我可以分享我们的代码样本。感谢您的帮助。2017-07-1110:45:07,580错误HSM|11.07.201710:45:07|HSMDecrypterroroccuredNet.Pkcs11Interop.Common.Pkcs

Java Stream流指南:优雅处理集合数据

文章目录一、为什么要使用stream流呢?二、如何获取Stream流?三、Stream流的中间方法四、Stream流的终结方法总结一、为什么要使用stream流呢?想必我们在日常编程中,会经常进行数据的处理,我们先来看看没有stram流时,我们的操作方式,我们想要收集姓赵的学生姓名。publicclassStreamDemo{publicstaticvoidmain(String[]args){ArrayListString>list=newArrayList>();Collections.addAll(list,"赵子龙","猪大肠","赵坤","张良","赵雯");ArrayListStr