使用Alpakka连接器的多个使用者线程 - java

我正在使用Alpakka kafka连接器来消耗来自kafka的数据包。我正在使用Consumer作为CommittableSource。我想在一台计算机上创建多个使用者线程，并将它们用作单个源。我该如何实现？

当前，我已经使用Consumer.CommittableSource创建了多个源，并使用“合并”功能将所有源合并为一个源。但是我不确定这是否是正确的方法，因为我没有创建线程。

请在下面找到我当前正在使用的源代码：

public Source<ConsumerMessage.CommittableMessage<String, String>, Consumer.Control> source() {
Source finalSource = Source.empty();
        for (int index = 0; index < consumerConfig.getNoOfConsumers(); index++) {
            finalSource = finalSource.merge(Consumer.committableSource(consumerSettings, subscription));
        }
return finalSource;
}

参考方案

是什么使您认为您需要更多的线程？您可能希望在多个流中共享单个Kafka消费者客户端实例。

您不应将来自多个Consumer.committableSource的元素合并到一个流中，它不适用于批量提交。

多次运行相同的流设置是否可以解决您的需求？

JAVA 8具有任何匹配属性的对象的过滤器列表 - java

我的要求是通过匹配任何属性的字符串来过滤对象列表。例如，假设Contact类具有三个属性：街道，城市，电话。我知道java流过滤器是如何工作的，在这里我必须将输入字符串与每个属性进行比较，如下所示：contactList.stream().filter(contact -> contact.getStreet().equals("dubai&…

java:继承 - java

有哪些替代继承的方法？ java大神给出的解决方案 有效的Java：偏重于继承而不是继承。 （这实际上也来自“四人帮”）。他提出的理由是，如果扩展类未明确设计为继承，则继承会引起很多不正常的副作用。例如，对super.someMethod()的任何调用都可以引导您通过未知代码的意外路径。取而代之的是，持有对本来应该扩展的类的引用，然后委托给它。这是与Eric…

Java-如何将此字符串转换为日期？ - java

我从服务器收到此消息，我不明白T和Z的含义，2012-08-24T09:59:59Z将此字符串转换为Date对象的正确SimpleDateFormat模式是什么？ java大神给出的解决方案 这是ISO 8601标准。您可以使用SimpleDateFormat simpleFormat = new SimpleDateFormat("yyyy-MM…

Java-固定大小的列表与指定初始容量的列表之间的差异 - java

我在理解这一点上遇到了问题。当我们做 List<Integer> list = Arrays.asList(array); 我们不能在该列表上使用添加，删除之类的方法。我知道Arrays.asList（）返回固定大小的列表。我不明白的是，如果我们创建一个具有指定初始容量的列表，例如List<Integer> list2 = new A…

从方法返回数组-Java - java

private static Coordinate[] getCircleCoordintaes() { Coordinate coordinates[] = {new Coordinate(0, 0)}; return coordinates; } 以上程序工作正常。在上面的程序中，返回的坐标数组首先初始化了数组使用这条线Coordinate coordi…