Pulsar IO: Sink/Source RabbitMq读写

本文介绍

近期需求要将RabbitMq 替换为 Pulsar, 所以要使用Pulsar IO 来将mq消息复制到Pulsar中并消费掉，或者Pulsar 发送消息复制到RabbitMq中消费掉。于是记录一下测试Pulsar消息复制到消费的全过程，使用的是Java代码。

Pulsar IO
官方概念：
Sources feed data from external systems into Pulsar.
Sinks feed data from Pulsar into external systems.

安装对应connector

$ wget https://archive.apache.org/dist/pulsar/pulsar-2.5.0/connectors/pulsar-io-rabbitmq-2.5.0.nar

$ mkdir connectors
$ mv pulsar-io-rabbitmq-2.5.0.nar$ ls connectors
pulsar-io-rabbitmq-2.5.0.nar

RabbitMQ source/sink connector

source配置官方文档
sink 配置官方文档
IO 使用官方文档

source

在./conf 文件夹下创建connectors-source.yml

configs:host: "192.168.1.22"port: 5672virtualHost: "/"username: "guest"password: "guest"queueName: "my-topic-source"connectionName: "test-rabbitmq-connection-source"requestedChannelMax: 0requestedFrameMax: 0connectionTimeout: 60000handshakeTimeout: 10000requestedHeartbeat: 60prefetchCount: 0prefetchGlobal: "false"

queueName 为队列topic 可在RabbitMq中配置

创建source

./bin/pulsar-admin sources create --archive connectors/pulsar-io-rabbitmq-2.5.0.nar --tenant public --namespace default --destination-topic-name my-topic --name pulsar-rabbitmq-source --source-config-file conf/connectors-source.yml

返回（sink同理）

"Created successfully"

相关参数

查看source状态（sink同理）

./bin/pulsar-admin sources status --tenant public --namespace default --name pulsar-rabbitmq-source

返回如下表示成功（sink同理）

{"numInstances" : 1,"numRunning" : 1,"instances" : [ {"instanceId" : 0,"status" : {"running" : true,"error" : "","numRestarts" : 0,"numReceivedFromSource" : 1,"numSystemExceptions" : 0,"latestSystemExceptions" : [ ],"numSourceExceptions" : 0,"latestSourceExceptions" : [ ],"numWritten" : 1,"lastReceivedTime" : 1596773830609,"workerId" : "c-pulsar-cluster-fw-192.168.1.214-8080"}} ]
}

查看log 文件路径

/tmp/functions/public/default/pulsar-rabbitmq-source/pulsar-rabbitmq-source-0.log

sink

在./conf 文件夹下创建connectors-sink.yml

configs:host: "192.168.1.22"port: 5672virtualHost: "/"username: "guest"password: "guest"queueName: "my-topic"connectionName: "test-rabbitmq-connection"requestedChannelMax: 0requestedFrameMax: 0connectionTimeout: 60000handshakeTimeout: 10000requestedHeartbeat: 60exchangeName: "amp.topic"routingKey: ""

创建sink

./bin/pulsar-admin sinks create --archive connectors/pulsar-io-rabbitmq-2.5.0.nar --tenant public --namespace default --inputs my-topic --name pulsar-rabbitmq-sink --sink-config-file conf/connectors-sink.yml

相关参数

发送消息

    // 发送Pulsarpublic void sendMsg() {
    Producer<byte[]> producer;try {
    producer = client.getPulsarClient().newProducer().topic("my-topic-source").create();// You can then send messages to the broker and topic you specified:producer.send("My message".getBytes());} catch (PulsarClientException e) {
    // TODO Auto-generated catch blocke.printStackTrace();}}// 发送Mq@Scheduled(fixedRate = Long.MAX_VALUE)public void sendMqMsg() throws PulsarClientException {
    messageClient.convertAndSend("my-topic", "ssssss");}

接收消息

public void sendMsg() throws PulsarClientException {
    consumer = client.getPulsarClient().newConsumer().topic("my-topic-source").subscriptionName("my-subscription").subscribe();Message msg = null;while (true) {
    // Wait for a messagetry {
    msg = consumer.receive();// Do something with the messageSystem.out.printf("Message received: %s", new String(msg.getData()));// Acknowledge the message so that it can be deleted by the message brokerconsumer.acknowledge(msg);} catch (Exception e) {
    // Message failed to process, redeliver laterSystem.out.println("error"+e.getCause());System.out.println("error"+e.getMessage());consumer.negativeAcknowledge(msg);}}}

使用指令消费topic

./bin/pulsar-client consume -s sub -n 0 my-topic-source

Pulsar接收RabbitMq 延迟消息

首先在mq中创建延迟 Exchange，创建好如下。

（未完待续）…
创建队列
将队列绑定到exchange，设置routing Key
设置延迟参数，发送消息。