rocketmq的消费者集群消费消息，实现负载均衡

package com.bfxy.rocketmq.model;

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.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import com.bfxy.rocketmq.constants.Const;

public class Consumer1 {

public Consumer1() {
try {
String group_name = "test_model_consumer_name";
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group_name);
consumer.setNamesrvAddr(Const.NAMESRV_ADDR_MASTER_SLAVE);
consumer.subscribe("test_model_topic2", "TagA");
consumer.setMessageModel(MessageModel.CLUSTERING);
//consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new Listener());
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}

class Listener implements MessageListenerConcurrently {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
try {
for(MessageExt msg : msgs){
String topic = msg.getTopic();
String msgBody = new String(msg.getBody(),"utf-8");
String tags = msg.getTags();
//if(tags.equals("TagA")) {
System.out.println("收到消息：" + " topic :" + topic + " ,tags : " + tags + " ,msg : " + msgBody);
//}
}
} catch (Exception e) {
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

}

public static void main(String[] args) {
Consumer1 c1 = new Consumer1();
System.out.println("c1 start..");

}
}

//==========================================================

package com.bfxy.rocketmq.model;

import java.util.List;

import com.bfxy.rocketmq.constants.Const;

public class Consumer2 {

public Consumer2() {
try {
String group_name = "test_model_consumer_name";
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group_name);
consumer.setNamesrvAddr(Const.NAMESRV_ADDR_MASTER_SLAVE);
consumer.subscribe("test_model_topic2", "TagA");
consumer.setMessageModel(MessageModel.CLUSTERING);
//consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new Listener());
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}

}

public static void main(String[] args) {
Consumer2 c2 = new Consumer2();
System.out.println("c2 start..");

}
}

原文地址：https://www.cnblogs.com/zhangzhiqin/p/10351557.html

时间： 2024-10-10 04:04:01

rocketmq的消费者集群消费消息，实现负载均衡

rocketmq的消费者集群消费消息，实现负载均衡的相关文章

集群、分布式、负载均衡区别与联系

分布式进阶(十九) 基于集群的动态反馈负载均衡策略

keepalived实现WEB服务集群的高可用负载均衡

集群、分布式、负载均衡

LVS集群中的IP负载均衡技术

使用ansible快速配置RHCS 集群实现WEB站负载均衡高可用（手记）

LVS集群十种调度算法及负载均衡理论

Linux集群之高可用负载均衡lvs+keepalived

集群之LVS（负载均衡）详解