共计 897 个字符,预计需要花费 3 分钟才能阅读完成。
语言:java(spring boot),单台 kafka
场景:同一个组,两个消费者同时生产一个 topic
实现过程:
首先批改这个 topic 的 partitions
./kafka-topics.sh --bootstrap-server localhost:9092 --alter --partitions 2 --topic topic-name
而后批改我的项目配置文件
spring:
kafka:
listener:
concurrency: 2
生产者代码
@SpringBootTest
class ProducerApplicationTests {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Test
void contextLoads() {for (int i = 1; i <= 1000; i++) {kafkaTemplate.send("test_0105_jcy", String.valueOf(i));
System.out.println(i);
}
}
}
消费者代码(X2)
@Slf4j
@SpringBootApplication
public class Customer1Application {@KafkaListener(topics = "test_0105_jcy")
public void consumerListener(String msg) throws InterruptedException {log.info(msg);
Thread.sleep(1 * 1000);
}
public static void main(String[] args) {SpringApplication.run(Customer1Application.class, args);
}
}
我的项目构造
生产者发送 1000 条音讯,两个消费者生产状况
依据默认策略将音讯发送到两个 partitions 中,如果有特殊要求的话,能够通过重载的 send 办法指定 partitions,send(String topic, Integer partition, K key, V data),也能够通过自定义分区器来实现。
正文完