/**
* 生产者
*/
public class Provider { public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException { //创建一个生产者 DefaultMQProducer producer=new DefaultMQProducer("rmq-group"); //设置NameServer地址 producer.setNamesrvAddr("192.168.7.11:9876;192.168.7.22:9876"); //设置生产者实例名称 producer.setInstanceName("provider"); //启动生产者 producer.start(); //发送消息 for (int i = 1; i <=10 ; i++) { Thread.sleep(1000); //模拟网络延迟 //创建消息 topic代表主题名称 tags代表小分类 body代表消息体 Message message=new Message("weksoft_topic","TagA",("wdksoft-"+i).getBytes()); //发送消息 SendResult send = producer.send(message); System.out.println(send.toString()); } } }
/**
* 消费者:监听消费
*/
public class Consumer { public static void main(String[] args) throws MQClientException { //创建消费者 DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group"); //设置NameServer地址 consumer.setNamesrvAddr("192.168.7.11:9876;192.168.7.22:9876"); //设置实例名称 consumer.setInstanceName("consumer"); //订阅Topic consumer.subscribe("weksoft_topic","TagA"); //监听消息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { //获取消息 for(MessageExt ext:msgs){ //RocketMQ由于是集群环境,所以产生的消息ID可能会重复 System.out.println(ext.getMsgId()+"----------"+new String(ext.getBody())); } //接受消息状态 1.消费成功 2.消费失败 队列还有 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); //启动消费者 consumer.start(); } }
2.RocketMQ重试机制
消费者重试:
报异常 int result = 5 / 0;
网络延迟
try { Thread.sleep(600000); } catch (InterruptedException e) { e.printStackTrace(); }
3.RocketMQ解决幂等性问题
//网络延迟,可能造成重复消费,如何解决重复消费(幂等性)问题
/**
* activeMQ:根据MessageID判断,获取当前的MessageID,判断和上一次是否一致,如果一致代表重复消费
* 如果不一致则进行消费
*
*
* RocketMQ:不能使用MessageID判断,因为在集群环境中,可能出现消息ID相同情况
* 消息Key,唯一,手动设置
*/
/* try { Thread.sleep(600000); } catch (InterruptedException e) { e.printStackTrace(); } */
3.1 生产者在生产数据时,指定数据的key,然后消费者进行数据消费时,获取到key,与redis中保存的key做判断,如果不相同
代表之前没有人进行消费,处理消费,保存到redis当中,当有第二个消费者时,如果拿到的消息与redis中相同代表之前已
已经有人消费。就进行数据签收,防止后续消费者同样拿到重复消费数据
3.2 生产者生产message指定key
public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException { //创建一个生产者 DefaultMQProducer producer=new DefaultMQProducer("rmq-group"); //设置NameServer地址 producer.setNamesrvAddr("192.168.7.11:9876;192.168.7.22:9876"); //设置生产者实例名称 producer.setInstanceName("provider"); //启动生产者 producer.start(); //发送消息 for (int i = 1; i <=1 ; i++) { Thread.sleep(1000); //模拟网络延迟 //创建消息 topic代表主题名称 tags代表小分类 body代表消息体 Message message=new Message("weksoft_topic_10","TagA",("wdksoft10-"+i).getBytes()); //消息的唯一标识 message.setKeys("订单编号"+i); //发送消息 SendResult send = producer.send(message); System.out.println(send.toString()); } }
3.3 消费者
public static void main(String[] args) throws MQClientException { //创建消费者 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group"); //设置NameServer地址 consumer.setNamesrvAddr("192.168.7.11:9876;192.168.7.22:9876"); //设置实例名称 consumer.setInstanceName("consumer"); //订阅Topic consumer.subscribe("weksoft_topic_10", "TagA"); //监听消息 consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { //获取消息 for (MessageExt ext : msgs) { //判断redis中有没有当前消息key if(map.containsKey(ext.getKeys())){ System.out.println("已经消费......."); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } //RocketMQ由于是集群环境,所以产生的消息ID可能会重复 System.out.println(ext.getMsgId() + "----------" + new String(ext.getBody())); //将当前Key保存到redis当中 map.put(ext.getKeys(),ext); } try{ int result=5/0; }catch (Exception e){ //人工补偿 return ConsumeConcurrentlyStatus.RECONSUME_LATER; } //接受消息状态 1.消费成功 2.消费失败 队列还有 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); //启动消费者 consumer.start(); }