我有一个需求,我需要消费Kafka Topic并将其写入MQ Topic。有人能建议我做它的最好的方法,我是新手卡夫卡。
我在confluent中读过关于IBM MQ Connector的文章,但不知道如何实现它。
发布于 2019-05-28 16:52:07
Kafka有一个叫做Kafka Connect的组件。它用于从Kafka读取数据和将数据写入其他系统,如您的MQ中的Database。
Kafka连接可以有两种类型的连接器
源连接器-从外部系统读取数据并写入Kafka (例如,从DB表中读取插入/修改的行,并在Kafka中插入到主题中)
接收器连接器-从Kafka读取消息写入外部系统。
您添加的链接是一个源连接器,它将从MQ读取消息并写入Kafka。
对于简单的用例,您不需要使用Kafka connect。您可以编写一个简单的Kafka consumer来读取Kafka主题中的数据并将其写入MQ。
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
//Insert code to append to MQ here
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
发布于 2019-05-28 16:35:45
将数据从Kafka移动到MQ的最佳方法是使用IBM MQ接收器连接器:https://github.com/ibm-messaging/kafka-connect-mq-sink
https://stackoverflow.com/questions/56337231
复制相似问题