RocketMQ知识
MQ基础概念：
rocketMQ：RocketMQ 是什么？Github 上关于 RocketMQ 的介绍：RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：
consumer group：1、概念：消费者分组，多个消费者在一个消费者分组中。2、注意点：一个consumer group中的机器相当于一个集群，consumer group中只有一台机器会接收到消息，并进行消费。每一个consumer group都会接收到消息。这样子的设计要求消费端需要保证幂等性。
topic：1、概念：Topic 是一种消息的逻辑分类，比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类，一个是订单 Topic 存放订单相关的消息，一个是库存 Topic 存储库存相关的消息。2、生产方发出的消息绑定某个topic，然后消费方监听某个topic，消费方（各个group）接收到消息，进行消费3、topic应用级别：整个应用最好都使用一个topic，而更加细的区分，使用tags来区分。
tag:1、概念：标签，用于对消息分类，在topic的基础上进行更细的划分。
nameServer：1、概念：Name Server 为 producer 和 consumer 提供路由信息。类似rpc中的注册中心。当producer需要发送消息首先去询问nameServer需要请求哪个broker。而当consumer需要拉取消息，也会先询问nameServer需要请求哪个broker。
broker:1、概念：rocketMQ中负责接收生产者消息、给消费者发送消息的组件。
Message：1、概念：Message 是消息的载体。一个 Message 必须指定 topic。Message 还有一个可选的 tag 设置，以便消费端可以基于 tag 进行过滤消息。也可以添加额外的键值对，例如你需要一个业务 key 来查找 broker 上的消息，方便在开发过程中诊断问题。
MQ生产者者实例：
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
//声明并初始化一个producer
//需要一个producer group名字作为构造方法的参数，这里为producer1
DefaultMQProducer producer = new DefaultMQProducer("producer1");
//设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
//NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
//调用start()方法启动一个producer实例
producer.start();
//发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
for (int i = 0; i < 10; i++) {
try {
Message msg = new Message("TopicTest",// topic
"TagA",// tag
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
);
//调用producer的send()方法发送消息
//这里调用的是同步的方式，所以会有返回结果
SendResult sendResult = producer.send(msg);
//打印返回结果，可以看到消息发送的状态以及一些相关信息
System.out.println(sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
//发送完消息之后，调用shutdown()方法关闭producer
producer.shutdown();
}
}MQ消费者实例：
在开发过程中，如果想测试生产者是否发出了mq，可以编写一个消费者进行测试
@Test
public void testMqConsumer() throws Exception {
String rocketmqAddress="10.113.41.2:9876;10.113.41.4:9876";
int threadNum = 5;
String topics = "WechatUnionCoreTemplateNotifyTopic";
String instanceName = "TemplateComsumer";
String groupName = "wechatUnionTemplateNotifyConsumer";
DefaultMQPushConsumer consumer = null;
consumer = new DefaultMQPushConsumer(groupName);
consumer.setNamesrvAddr(rocketmqAddress);//MQ地址
consumer.setClientCallbackExecutorThreads(threadNum);//消费现场数量
consumer.setInstanceName(instanceName);//实例名称
consumer.subscribe(topics, "*");
//注册监听
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (int i = 0; i < msgs.size(); i++) {
MessageExt msgExt = msgs.get(i);
String msgId = msgExt.getMsgId();
Integer flag = msgExt.getFlag();
TemplateNotifyItem templateNotifyItem = ProtoBufSerialize.fromProto(msgExt.getBody(), TemplateNotifyItem.class);
logger.info("receive new Msg: " + " msgId=" + msgId + " flag=" + flag + " templateNotifyItem=" + templateNotifyItem);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
logger.info("监听执行中");
Thread.sleep(1000000);
}参考： http://blog.csdn.net/manzhizh... https://www.jianshu.com/p/824... 架构师之路
腾讯云开发者
