RocketMQ角色

集群特点

NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步Broker部署相对复杂，Broker分为Master与Slave。一个master可以对应多个slave，但是一个slave只能对应一个master，master与slave的对应关系通过指定相同的brokername，不同的brokerid来定义。brokerId为0表示master，非0表示slave。master可以部署多个，每个broker与nameserver集群中的所有节点建立长连接，定时注册topic信息到所有nameserver，并且有broker向nameserver汇报自己的存活状态producer与nameserver集群中的其中一个节点（随机选择）建立长连接，定期从nameserver取topic路由信息，并向提供topic服务的master建立长连接，且定时向master发送心跳。producer完全无状态，可集群部署Consumer与nameserver集群中的其中一个节点（随机选择）建立长连接，定期从nameserver取topic路由信息，并向提供topic服务的master，salve建立长连接，且定时向master，slave发送心跳。consumer既可以从master订阅消息，也可以从slave订阅消息，订阅规则由broker配置决定

集群模式

1）单Master模式

这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。

2）多Master模式

一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：

优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。

3）多Master多Slave模式（异步）

每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；缺点：Master宕机，磁盘损坏情况下会丢失少量消息。

4）多Master多Slave模式（同步）

每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

搭建环境(windows)

1.系统环境变量配置

变量名：ROCKETMQ_HOME 变量值：MQ解压路径\MQ文件夹名

2. 启动

首先启动nameserver，Cmd命令框执行进入至‘MQ文件夹\bin’下，然后执行‘start mqnamesrv.cmd’，启动NAMESERVER再启动broker， Cmd命令框执行进入至‘MQ文件夹\bin’下，然后执行‘start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true’，启动BROKER更多详细 https://blog.csdn.net/kingtok/article/details/104212625

消息发送

1.同步消息发送

producer向 broker 发送消息，执行 API 时同步等待， 直到broker 服务器返回发送结果

//1.创建消息生产者producer,并制定生产者组名 DefaultMQProducer producer = new DefaultMQProducer("group1"); //2.指定Nameserver地址 producer.setNamesrvAddr("127.0.0.1:9876"); //3.启动producer producer.start(); for (int i = 0;i<10;i++){ //4.创建消息对象，指定主题Topic,Tag和消息体 /** * 参数1: 消息主题Topic * 参数2：消息Tag * 消息3：消息内容 */ Message msg = new Message("base","Tag1",("Hello World"+i).getBytes()); //5.发送消息 SendResult result = producer.send(msg); System.out.println("消息结果："+result); Thread.sleep(1000); } producer.shutdown();

2.异步消息发送

producer向 broker 发送消息时指定消息发送成功及发送异常的回调方法，调用 API 后立即返回，producer发送消息线程不阻塞 ，消息发送成功或失败的回调任务在一个新的线程中执行 。

//1.创建消息生产者producer,并制定生产者组名 DefaultMQProducer producer = new DefaultMQProducer("group1"); //2.指定Nameserver地址 producer.setNamesrvAddr("127.0.0.1:9876"); //3.启动producer producer.start(); for (int i = 0;i<10;i++){ //4.创建消息对象，指定主题Topic,Tag和消息体 /** * 参数1: 消息主题Topic * 参数2：消息Tag * 消息3：消息内容 */ Message msg = new Message("base","Tag2",("Hello World"+i).getBytes()); //5.发送异步消息 producer.send(msg, new SendCallback() { //发送成功的回调函数 @Override public void onSuccess(SendResult sendResult) { System.out.println("发送结果："+sendResult); } //发送失败的回调函数 @Override public void onException(Throwable throwable) { System.out.println("异常"); } }); Thread.sleep(1000); } producer.shutdown();

3.单向消息发送

producer向 broker 发送消息，执行 API 时直接返回，不等待broker 服务器的结果 。 sendOneway();

//1.创建消息生产者producer,并制定生产者组名 DefaultMQProducer producer = new DefaultMQProducer("group1"); //2.指定Nameserver地址 producer.setNamesrvAddr("127.0.0.1:9876"); //3.启动producer producer.start(); for (int i = 0;i<10;i++){ //4.创建消息对象，指定主题Topic,Tag和消息体 /** * 参数1: 消息主题Topic * 参数2：消息Tag * 消息3：消息内容 */ Message msg = new Message("base","Tag1",("Hello World 单向消息"+i).getBytes()); //5.发送消息 producer.sendOneway(msg); Thread.sleep(1000); } producer.shutdown();

消费者模式

负载均衡模式：默认的消费模式广播模式：consumer.setMessageModel(MessageModel.BROADCASTING); //1.创建消费者Consumer,制定消费者组名 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1"); //2.指定Nameserver地址 consumer.setNamesrvAddr("127.0.0.1:9876"); //3.订阅主题Topic和Tag consumer.subscribe("base","Tag1"); //设置消费模式：负载均衡||广播模式，默认负载均衡 consumer.setMessageModel(MessageModel.BROADCASTING);//广播 //4.设置回调函数，处理消息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { for (MessageExt msg:list) { System.out.println(new String(msg.getBody())); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); //5.启动消费者consumer consumer.start();

顺序

Producer：需要设置队列标识 public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException { //1.创建消息生产者producer,并制定生产者组名 DefaultMQProducer producer = new DefaultMQProducer("group1"); //2.指定Nameserver地址 producer.setNamesrvAddr("127.0.0.1:9876"); //3.启动producer producer.start(); //构建消息集合 List<OrderStep> orderSteps = OrderStep.buildOrders(); //发送消息 int i = 0; for (OrderStep order:orderSteps) { Message message = new Message("OrderTopic","Order","i"+i,order.toString().getBytes()); i = i+1; message.setDelayTimeLevel(2);//延迟 /** * 参数1：消息队列 * 参数2：消息队列的选择器 * 参数3：选择队列的业务标识（订单ID） */ SendResult sendResult = producer.send(message, new MessageQueueSelector() { /** * @param list 队列集合 * @param message 消息对象 * @param o 业务标识的参数 * @return */ @Override public MessageQueue select(List<MessageQueue> list, Message message, Object o) { long orderId = (long) o; long index = orderId%list.size(); System.out.println(index+"~~~"+orderId); return list.get((int) index); } },order.getOrderId()); System.out.println("发送结果："+sendResult); // Thread.sleep(1000); } producer.shutdown(); } Consumer：顺序需要使用MessageListenerOrderly public static void main(String[] args) throws MQClientException { //1.创建消费者Consumer,制定消费者组名 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1"); //2.指定Nameserver地址 consumer.setNamesrvAddr("127.0.0.1:9876"); //3.订阅主题Topic和Tag consumer.subscribe("OrderTopic","*"); //4.设置回调函数，处理消息 consumer.registerMessageListener(new MessageListenerOrderly() { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) { for (MessageExt msg:list) { System.out.println("线程名称："+Thread.currentThread().getName()+"接受消息："+ new String(msg.getBody())+"队列："+msg.getQueueId() +"，延迟："+(System.currentTimeMillis()-msg.getBornTimestamp())); } return ConsumeOrderlyStatus.SUCCESS; } }); //5.启动消费者consumer consumer.start(); System.out.println("消费者启动"); }

延迟

将发送的消息进行设置，需要注意的是这里传入的参数不是延迟几秒，而是延迟的等级 messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 如：1就代表1s，2就代表5s。

message.setDelayTimeLevel(2);//延迟

批量

将消息装入集合发送即可

过滤消息

设置consumer

tags consumer.subscribe("base","Tag1 || Tag2");//消费多个tag通过||设置 sql consumer.subscribe("base", MessageSelector.bySql("i>4"));

事务

consumer不变 producer：设置事务监听器，需要注意的是不能把producer.shutdown()了，不然mq进行消息回查不行

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException { //1.创建消息生产者producer,并制定生产者组名 TransactionMQProducer producer = new TransactionMQProducer("group1"); //2.指定Nameserver地址 producer.setNamesrvAddr("127.0.0.1:9876"); //事务监听器 producer.setTransactionListener(new TransactionListener() { //在该方法中执行本地事务 @Override public LocalTransactionState executeLocalTransaction(Message message, Object o) { if (message.getTags().equals("Tag1")){ return LocalTransactionState.COMMIT_MESSAGE; }else if (message.getTags().equals("Tag2")){ return LocalTransactionState.ROLLBACK_MESSAGE; }else if (message.getTags().equals("Tag3")){ return LocalTransactionState.UNKNOW; } return LocalTransactionState.UNKNOW; } //该方法是MQ进行消息事务状态回差 @Override public LocalTransactionState checkLocalTransaction(MessageExt messageExt) { System.out.println("消息回滚"); return LocalTransactionState.COMMIT_MESSAGE; } }); //3.启动producer producer.start(); //构建消息集合 List<Message> list = new ArrayList<>(); Message msg1 = new Message("base","Tag1",("Hello World 1").getBytes()); msg1.putUserProperty("i","3"); Message msg2 = new Message("base","Tag2",("Hello World 2").getBytes()); msg2.putUserProperty("i","4"); Message msg3 = new Message("base","Tag3",("Hello World 3").getBytes()); msg3.putUserProperty("i","5"); list.add(msg1); list.add(msg2); list.add(msg3); //发送消息 producer.sendMessageInTransaction(msg1,null); producer.sendMessageInTransaction(msg2,null); producer.sendMessageInTransaction(msg3,null); // Thread.sleep(1000); // producer.shutdown(); }