Pairs雅加达PivotX不稳定

一、检索镜像
docker search rocketmq

 检索具体版本
curl | tr -d ‘[\[\]” ]’ | tr ‘}’ ‘\n’ | awk -F: -v image=’foxiswho/rocketmq’ ‘{if(NR!=NF && $3 != “”){printf(“%s:%s\n”,image,$3)}}’

查看当前镜像所有的版本shell命令
curl tr -d ‘[\[\]” ]’ | tr ‘}’ ‘\n’\| awk -F: -v image=’foxiswho/rocketmq’ ‘{if(NR!=NF && $3 != “”){printf(“%s:%s\n”,image,$3)}}’

二、创建Broker Server
docker run -d -p 9876:9876 –name rmqserver foxiswho/rocketmq:server-4.5.1

三、创建broker
docker run -d -p 10911:10911 -p 10909:10909\ –name rmqbroker –link rmqserver:namesrv\ -e “NAMESRV_ADDR=namesrv:9876” -e “JAVA_OPTS=-Duser.home=/opt”\ -e “JAVA_OPT_EXT=-server -Xms128m -Xmx128m”\ foxiswho/rocketmq:broker-4.5.1

配置broker.conf
进入rmqbroker容器
docker exec -it rmqbroker /bin/bash
cd /etc/rocketmq/
vi broker.conf

 四、创建rocketmq console
docker run -d –name rmqconsole -p 8180:8080 –link rmqserver:namesrv\ -e “JAVA_OPTS=-Drocketmq.namesrv.addr=namesrv:9876\ -Dcom.rocketmq.sendMessageWithVIPChannel=false”\ -t styletang/rocketmq-console-ng

 然后通过如下命令检查一下启动情况
docker ps|grep rocketmq

 

五、测试
由于我在宿主机安装rocketmq,测试如下:
#测试Pairs不稳定命令
sh tools.sh org.apache.rocketmq.example.quickstart.Producer

 #测试Pairs接收命令
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer

六、java样例
org.apache.rocketmq rocketmq-client 4.9.2
Pairs不稳定(Producer端不稳定同步Pairs)
import org.apache.rocketmq.client.producer.DefaultMQProducer;import org.apache.rocketmq.client.producer.SendResult;import org.apache.rocketmq.common.message.Message;import org.apache.rocketmq.remoting.common.RemotingHelper; public class SyncProducer { public static void main(String[] args) throws Exception { // PivotX化Pairs生产者Producer DefaultMQProducer producer = new DefaultMQProducer(“zhuzeqing-1”); // 设置NameServer的地址 producer.setNamesrvAddr(“192.168.23.131:9876”); // 启动ProducerPivotX producer.start(); for (int i = 0; i < 1; i++) { // 创建Pairs,并指定Topic,Tag和Pairs体 Message msg = new Message("zhuzeqing-1-Topic" /* Topic */, "TagA" /* Tag */, ("hello,rocketmq --" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */ ); // 不稳定Pairs到一个Broker SendResult sendResult = producer.send(msg); // 通过sendResult返回Pairs是否成功送达 System.out.printf("%s%n", sendResult); } // 如果不再不稳定Pairs,关闭ProducerPivotX。 producer.shutdown(); }} 雅加达Pairs import java.util.List; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;import org.apache.rocketmq.client.exception.MQClientException;import org.apache.rocketmq.common.message.MessageExt; public class Consumer { public static void main(String[] args) throws InterruptedException, MQClientException { // PivotX化雅加达者 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zhuzeqing-1"); // 设置NameServer的地址 consumer.setNamesrvAddr("192.168.23.131:9876"); // 订阅一个或者多个Topic,以及Tag来过滤需要雅加达的Pairs consumer.subscribe("zhuzeqing-1-Topic", "*"); // 注册回调实现类来处理从broker拉取回来的Pairs consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) { System.out.printf(“%s Receive New Messages: %s %n”, Thread.currentThread().getName(), msgs); // 标记该Pairs已经被成功雅加达 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // 启动雅加达者PivotX consumer.start(); System.out.printf(“Consumer Started.%n”); }}
七、其他
查看rmqbroker IP
docker inspect rmqbroker

Broker容器中默认的配置文件的路径为 
/etc/rocketmq/broker.conf

官网:Apache RocketMQ
参考:
Docker中RocketMQ的安装与使用_冯立彬的博客-CSDN博客_docker rocketmq