Как потреблять Кафку Тема в Consumer MQ Тема - PullRequest
Новая
       19

Как потреблять Кафку Тема в Consumer MQ Тема

0 голосов
/ 28 мая 2019

У меня есть требование, где мне нужно использовать тему Кафки и записать ее в тему MQ. Может кто-нибудь посоветовать мне лучший способ сделать это, я новичок в Кафке.

Я читал о IBM MQ Connector в слиянии, но не смог понять, как его реализовать.

Ответы [ 2 ]

1 голос
/ 28 мая 2019

Лучший способ переместить данные из Kafka в MQ - использовать соединитель приемника IBM MQ: https://github.com/ibm-messaging/kafka-connect-mq-sink

Это разъем Kafka Connect. README содержит сведения о сборке и работе it.

0 голосов
/ 28 мая 2019

Kafka имеет компонент под названием Kafka Connect.Он используется для чтения и записи данных в / из Kafka в другие системы, такие как База данных в вашем случае MQ.

Kafka connect может иметь два типа разъемов

Исходные разъемы - считывать данные извнешней системы и записи в Kafka (например, для чтения вставленных / измененных строк из таблицы в БД и вставки в тему в Kafka)

Sink Connector - чтение сообщения из записи Kafka во внешнюю систему.

Ссылка, которую вы добавили, является исходным соединителем, она будет читать сообщения из MQ и записывать в Kafka.

Для простого случая использования вам не нужно подключать Kafka.Вы можете написать простой потребитель Kafka , который будет читать данные из темы 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());
     }
...