RocketMQ入门-搭建与demo

Apache RocketMQ由4个部分组成：nameServer，broker，producer，consumer，另外还提供了一个console用于监控。

• nameServer会提供基本的服务发现和路由，producer和consumer通过nameServer可以获取到broker的地址，从而与broker进行通讯
• broker则提供真正的MQ功能
  
  图片来自官网http://rocketmq.apache.org/docs/rmq-arc/

使用docker搭建RocketMQ

首先确认机器的内网地址和外网地址，我以{内网IP}，{外网IP}表示

nameserver

docker run -d -p 9876:9876 --name rmqserver \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
foxiswho/rocketmq:server

broker

在本地创建配置文件/etc/rocketmq/broker.conf

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
# 以下是我添加的
# 配置nameserver的地址，broker需要上报状态给nameserver
namesrvAddr = {内网IP}:9876
# 配置监听地址，nameserver会将此地址提供给客户端使用
# 如果mq服务需要在外网使用则配成外网，需要在内网使用则配成内网地址
brokerIP1 = {外网IP}

docker run -d -p 10911:10911 -p 10909:10909 --name rmqb \
-v /etc/rocketmq/broker.conf:/etc/rocketmq/broker.conf \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
foxiswho/rocketmq:broker

console

docker run --name rmqconsole -p 8080:8080 \
-e "JAVA_OPTS=-Drocketmq.namesrv.addr={内网IP}:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" \
-t styletang/rocketmq-console-ng

demo

消费者

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("s1");
consumer.setNamesrvAddr("{外网IP}:9876");
consumer.subscribe("mytopic", "mytag");
consumer.registerMessageListener(new MessageListenerConcurrently() {
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		for(int i=0; i<msgs.length(); i++){
			MessageExt msg = msgs.get(i);
			System.out.println(msg.getTopic() + " " + msg.getTags() + " " + new String(msg.getBody()));
		}
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}
});
consumer.start();

生产者

DefaultMQProducer producer = new DefaultMQProducer("s2");
producer.setNamesrvAddr("{外网IP}:9876");
producer.start();
Message msg = new Message();
msg.setTopic("mytopic");
msg.setTags("mytag");
msg.setBody("mybody".getBytes());
producer.send(msg);