我开始在我的项目中使用spring-integration-kafka,我可以生产和消费来自kafka的消息。但是现在,我想为特定的分区生成消息,并从特定的分区消费消息。
例如,我想向分区3生成消息,而消费将只接收来自分区3的消息。
到目前为止,我的主题有8个分区,我可以向特定的分区发送消息,但是我还没有找到配置消费者只接收来自特定分区的消息的方法。
因此,任何关于我应该如何配置消费者与sping-Integration-kafka的建议,或任何其他需要与KafkaConsumer.java类做的事情,因为它可以接收来自特定分区的消息。
谢谢。
这是我的代码:
kafka-producer-context.xml
<int:publish-subscribe-channel id="inputToKafka" />
<int-kafka:outbound-channel-adapter
id="kafkaOutboundChannelAdapter" kafka-producer-context-ref="kafkaProducerContext"
auto-startup="true" order="1" channel="inputToKafka" />
<int-kafka:producer-context id="kafkaProducerContext"
producer-properties="producerProps">
<int-kafka:producer-configurations>
<int-kafka:producer-configuration
broker-list="127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094"
async="true" topic="testTopic"
key-class-type="java.lang.String"
key-encoder="encoder"
value-class-type="java.lang.String"
value-encoder="encoder"
partitioner="partitioner"
compression-codec="default" />
</int-kafka:producer-configurations>
</int-kafka:producer-context>
<util:properties id="producerProps">
<prop key="queue.buffering.max.ms">500</prop>
<prop key="topic.metadata.refresh.interval.ms">3600000</prop>
<prop key="queue.buffering.max.messages">10000</prop>
<prop key="retry.backoff.ms">100</prop>
<prop key="message.send.max.retries">2</prop>
<prop key="send.buffer.bytes">5242880</prop>
<prop key="socket.request.max.bytes">104857600</prop>
<prop key="socket.receive.buffer.bytes">1048576</prop>
<prop key="socket.send.buffer.bytes">1048576</prop>
<prop key="request.required.acks">1</prop>
</util:properties>
<bean id="encoder"
class="org.springframework.integration.kafka.serializer.common.StringEncoder" />
<bean id="partitioner" class="org.springframework.integration.kafka.support.DefaultPartitioner"/>
<task:executor id="taskExecutor" pool-size="5"
keep-alive="120" queue-capacity="500" />
KafkaProducer.java
public class KafkaProducer {
private static final Logger logger = LoggerFactory
.getLogger(KafkaProducer.class);
@Autowired
private MessageChannel inputToKafka;
public void sendMessage(String message) {
try {
inputToKafka.send(MessageBuilder.withPayload(message)
.setHeader(KafkaHeaders.TOPIC, "testTopic")
.setHeader(KafkaHeaders.PARTITION_ID, 3).build());
} catch (Exception e) {
logger.error(String.format(
"Failed to send [ %s ] to topic %s ", message, topic),
e);
}
}
}
kafka-consumer-context.xml
<int:channel id="inputFromKafka">
<int:dispatcher task-executor="kafkaMessageExecutor" />
</int:channel>
<int-kafka:zookeeper-connect id="zookeeperConnect"
zk-connect="127.0.0.1:2181" zk-connection-timeout="6000"
zk-session-timeout="6000" zk-sync-time="2000" />
<int-kafka:inbound-channel-adapter
id="kafkaInboundChannelAdapter" kafka-consumer-context-ref="consumerContext"
auto-startup="true" channel="inputFromKafka">
<int:poller fixed-delay="10" time-unit="MILLISECONDS"
max-messages-per-poll="5" />
</int-kafka:inbound-channel-adapter>
<bean id="consumerProperties"
class="org.springframework.beans.factory.config.PropertiesFactoryBean">
<property name="properties">
<props>
<prop key="auto.offset.reset">smallest</prop>
<prop key="socket.receive.buffer.bytes">1048576</prop>
<prop key="fetch.message.max.bytes">5242880</prop>
<prop key="auto.commit.interval.ms">1000</prop>
</props>
</property>
</bean>
<int-kafka:consumer-context id="consumerContext"
consumer-timeout="1000" zookeeper-connect="zookeeperConnect"
consumer-properties="consumerProperties">
<int-kafka:consumer-configurations>
<int-kafka:consumer-configuration
group-id="defaultGrp" max-messages="20000">
<int-kafka:topic id="testTopic" streams="3" />
</int-kafka:consumer-configuration>
</int-kafka:consumer-configurations>
</int-kafka:consumer-context>
<task:executor id="kafkaMessageExecutor" pool-size="0-10"
keep-alive="120" queue-capacity="500" />
<int:outbound-channel-adapter channel="inputFromKafka"
ref="kafkaConsumer" method="processMessage" />
KafkaConsumer.java
public class KafkaConsumer {
private static final Logger log = LoggerFactory
.getLogger(KafkaConsumer.class);
@Autowired
KafkaService kafkaService;
public void processMessage(Map<String, Map<Integer, List<byte[]>>> msgs) {
for (Map.Entry<String, Map<Integer, List<byte[]>>> entry : msgs
.entrySet()) {
log.debug("Topic:" + entry.getKey());
ConcurrentHashMap<Integer, List<byte[]>> messages = (ConcurrentHashMap<Integer, List<byte[]>>) entry
.getValue();
log.debug("\n**** Partition: \n");
Set<Integer> keys = messages.keySet();
for (Integer i : keys)
log.debug("p:"+i);
log.debug("\n**************\n");
Collection<List<byte[]>> values = messages.values();
for (Iterator<List<byte[]>> iterator = values.iterator(); iterator
.hasNext();) {
List<byte[]> list = iterator.next();
for (byte[] object : list) {
String message = new String(object);
log.debug("Message: " + message);
try {
kafkaService.receiveMessage(message);
} catch (Exception e) {
log.error(String.format("Failed to process message %s",
message));
}
}
}
}
}
}
所以我的问题就在这里。当我向分区 3 或任何分区生成消息时,Kafka 消费者始终会收到该消息。我想要的只是:Kafka消费者只会从分区3接收消息,而不是从其他分区接收消息。
再次感谢。
您需要使用消息驱动通道适配器。
作为变体,KafkaMessageListenerContainer可以接受org . spring framework . integration . Kafka . core . partition数组参数来指定主题及其分区对。
您需要使用此构造函数连接侦听器容器,并使用侦听器容器
属性将其提供给适配器。
我们将用一个例子来更新自述文件。
我有一个名为“test-topic”的主题,有3个分区。 当我启动一个将group-id设置为“test-group”的使用者(consumer-1)时,它连接并读取主题上的所有分区。到目前为止还好。 当我在同一个组中启动另一个消费者(consumer-2)时,问题就出现了。我希望在两个消费者之间划分分区时能够重新平衡,例如,消费者-1得到分区0和2,消费者-2得到分区1。这种情况不会发生,当然我
我不知道是怎么回事,我的java客户机消费者用@KafkaListener注释后没有收到任何消息。当我通过命令行创建消费者时,它可以工作。同样,Producer也能按预期工作(同样在java中)。有人能帮我理解这种行为吗? application.yml 生产者配置: 消费者配置: 制作人 Spring控制器: 这是我的控制台输出,正如您所看到的,它发送一条消息,但该方法不接收任何内容。如果我没有
我们希望在读取消息表单kafka时实现并行性。因此我们想在flinkkafkaconsumer中指定分区号。它将从kafka中的所有分区读取消息,而不是特定的分区号。以下是示例代码: 请建议任何更好的选择来获得并行性。
我在使用时遇到了困难,无法从开头或其他任何显式偏移量读取它。 为同一主题的使用者运行命令行工具,我确实看到带有选项的消息,否则它将挂起 我使用的是kafka-python 0.9.5,而代理运行的是Kafka8.2。不确定确切的问题是什么。 按照dpkp的建议设置_group_id=none_以模拟控制台使用者的行为。
本文向大家介绍Kafka 消费者是否可以消费指定分区消息?相关面试题,主要包含被问及Kafka 消费者是否可以消费指定分区消息?时的应答技巧和注意事项,需要的朋友参考一下 Kafa consumer消费消息时,向broker发出fetch请求去消费特定分区的消息,consumer指定消息在日志中的偏移量(offset),就可以消费从这个位置开始的消息,customer拥有了offset的控制权,可
我知道kafka将一个主题的数据安排在许多分区上,一个消费者组中的消费者被分配到不同的分区,从那里他们可以接收数据: 我的问题是: 术语,它们是由主机/IP标识的,还是由客户端连接标识的? 换句话说,如果我启动两个线程或进程,使用相同的消费者组运行相同的Kafka客户端代码,它们被认为是一个消费者还是两个消费者?