消息队列：Message Queue，MQ
历史：阿里巴巴于2012年开源了RocketMQ，2016年贡献给了Apache基金会
官网：http://rocketmq.apache.org/
常用MQ对比
Apache ActiveMQ：不够轻巧，对于队列较多的情况支持不好，还有丢消息的情况
Apache Apollo：是ActiveMQ的下一代产品，目前市场接收度不够
Apache Kafka：高性能，高可靠
Apache RocketMQ：思路源于Kafka，并做了优化
RabbitMQ：性能较差
ZeroMQ：性能高，但不支持消息持久化，不保证可靠性
角色：消息生成者Producer，消息消费者Consumer，消息中间人Broker，服务路由器NameServer
概念：消息主体（Topic）
异步的意义：压力缓冲，流量削峰，降低主流程耗时，较少主流程出错
典型应用
微信小程序服务消息推送，单次消费机制实现单次推送，延迟消费机制实现定时推送
Windows 安装与配置
1、下载bin.zip，解压
2、设置环境变量ROCKETMQ_HOME=解压根目录
3、bin目录下，编辑runserver.cmd和runbroker.cmd，内存配置改为-Xms512m -Xmx512m -Xmn256m
4、bin目录下，启动NameServer：start mqnamesrv.cmd
5、启动Broker：start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true
Linux安装与配置
安装步骤
1、下载 rocketmq-all-.zip，上传到 /usr
2、解压 unzip rocketmq-all-.zip
3、启动NameServer：sh bin/mqnamesrv
4、配置conf/broker.conf，祥见下文
5、启动Broker：sh bin/mqbroker -c rocketmq/conf/broker.conf
部署策略
1、安装两个以上对等的NameServer
2、安装多个Broker-Master
3、安装多个Broker-Master对应的Broker-Slave
conf/broker.conf
brokerClusterName=bokers # 集群名称，一组NameServer可以服务于多个Broker集群
brokerName=broker-a # 一组主从Broker，名称要一样
brokerId=1 # 本机在这组主从Broker中的编号
brokerRole=SLAVE # 本机在这组主从Broker中的角色，ASYNC_MASTER或SLAVE
namesrvAddr=192.168.6.3:9876;192.168.6.4:9876 # NameServer的地址，用于broker找到nameserver，如果用公网IP得保证，nameserver端口公网可访问
brokerIP1=192.168.6.3 # 本机IP，用于登记在nameserver，如果用公网IP得保证，broker端口公网可访问
listenPort=10920 # 监听端口
autoCreateTopicEnable=false # 是否自动创建Topic
RocketMQ命令
查看集群状态：sh bin/mqadmin clusterList -n NameServer地址 # 可在任意节点上执行
创建Topic：sh bin/mqadmin updateTopic -c 集群名称 -n "NameServer地址列表" -t Topic1 -r 4 -w 4
消息管理
清空消息：RocketMQ管理界面里，删除topic 并 更改Consumer Group（已经分配的消息会积压在原Consumer Group）
RocketMQ应用组成
1、配置Maven依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.3.1</version>
</dependency>
2、生产者
DefaultMQProducer producer = new DefaultMQProducer("group_1"); // 传入生产者组的名称
try {
producer.setNamesrvAddr("192.168.6.3:9876;192.168.6.4:9876"); // 传入NameServer地址
producer.setInstanceName(UUID.randomUUID().toString());
producer.setVipChannelEnabled(false);
producer.start();
Message msg = new Message("topic_1", "hello rocketmq".getBytes()); // 传入主题和消息
message.setDelayTimeLevel(3); // 设置延迟级别，可以用于实现固定时间点任务
// 延迟级别列表默认为"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"，可以在Broker的配置文件中配置
producer.send(msg);
}finally {
producer.shutdown();
}
3、消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_1"); // 传入消费者组名称
consumer.setNamesrvAddr("192.168.6.3:9876;192.168.6.4:9876"); // 传入NameServer地址
consumer.setInstanceName(UUID.randomUUID().toString());
consumer.setConsumeMessageBatchMaxSize(32);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerConcurrently() { // 注册消息处理器
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for(MessageExt me : list) {
System.out.print(new String(me.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.subscribe("topic_1", "*"); // 监听Topic
consumer.start();
Spring整合
1、监听器
public class ConsumerImpl implements MessageListenerConcurrently {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
//CONSUME_SUCCESS 消费成功
//RECONSUME_LATER 消费失败
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
2、Spring配置文件
<!-- 生成者 -->
<bean id="rocketmqProducer" class="org.apache.rocketmq.client.producer.DefaultMQProducer" init-method="start" destroy-method="shutdown">
<property name="producerGroup" value="group_1"/>
<property name="namesrvAddr" value="127.0.0.1:9876"/>
</bean>
<!-- 监听器 -->
<bean id="consumerImpl" class="com.hogen.consumer.ConsumerImpl" />
<!-- 消费者 -->
<bean id="rocketmqConsumer" class="org.apache.rocketmq.client.consumer.DefaultMQPushConsumer" init-method="start" destroy-method="shutdown">
<property name="consumerGroup" value="concurrent_consumer"/>
<property name="namesrvAddr" value="127.0.0.1:9876"/>
<property name="messageListener" ref="consumerImpl"/>
<property name="subscription">
<map>
<entry key="Topic_1">
<value>*</value>
</entry>
</map>
</property>
</bean>
SpringBoot 整合
1、配置Maven依赖
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.5.2</version>
</dependency>
2、ProducerConfig
@Configuration
public class ProducerConfig {
private String namesrvAddr = "127.0.0.1:9876";
private String groupName = "DefaultCluster";
@Bean
@ConditionalOnProperty(prefix = "rocketmq.producer", value = "default", havingValue = "true")
public DefaultMQProducer defaultProducer() throws MQClientException {
DefaultMQProducer producer = new DefaultMQProducer(this.groupName);
producer.setNamesrvAddr(this.namesrvAddr );
producer.setVipChannelEnabled(false);
producer.setRetryTimesWhenSendAsyncFailed(10);
producer.start();
return producer;
}
}
3、使用 producer
@Autowired
private DefaultMQProducer producer;
Message message = new Message("topic", "tags", "key", "内容".getBytes());
producer.send(message, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
logger.info("MQ传输成功");
}
@Override
public void onException(Throwable e) {
logger.error("MQ传输失败", e);
}
});
4、customer
@Configuration
public class MqCustomerConfig implements ApplicationListener<ContextRefreshedEvent> {
private String namesrvAddr = "127.0.0.1:9876";
private String groupName = "DefaultCluster";
@Override
public void onApplicationEvent(ContextRefreshedEvent arg0) {
try {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.groupName);
consumer.setNamesrvAddr(this.namesrvAddr);
consumer.subscribe("topic", "tag");
// 开启内部类实现监听
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
return MqCustomerConfig.this.dealBody(msgs);
}
});
consumer.start();
} catch (MQClientException e) {
// 消费者监听器启动失败
}
}
public ConsumeConcurrentlyStatus dealBody(List<MessageExt> msgs) {
for(MessageExt msg : msgs) {
try {
String msgStr = new String(msg.getBody(), "utf-8");
} catch (UnsupportedEncodingException e) {
// body转字符串解析失败
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
阿里云
消息队列 Kafka
消息队列 RocketMQ 即 ONS（Open Notification Service）
消息服务 MNS（Message Notification Service）原 MQS（Message Queue Service），阿里云上特有组件，性能比ONS高
SpringBoot 集成 阿里云RocketMQ
1、pom.xml
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>1.8.7.1.Final</version>
</dependency>
2、Consumer 和 Producer 的bean
@Configuration
public class MqConfig {
@Autowired
private MqHandler mqHandler;
@Autowired
@Qualifier("MqProperties")
private Properties properties;
@Bean("MqProperties")
public Properties properties(){
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.AccessKey, accessKey);
properties.setProperty(PropertyKeyConst.SecretKey, secretKey);
properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, ONSAddr); // TCP 地址
properties.setProperty(PropertyKeyConst.GROUP_ID, groupId);
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
return properties;
}
@Bean(initMethod = "start", destroyMethod = "shutdown")
public Consumer consumer(){
Consumer consumer = ONSFactory.createConsumer(properties);
consumer.subscribe("game_news", "*", mqHandler);
return consumer;
}
@Bean(initMethod = "start", destroyMethod = "shutdown")
public Producer producer(){
Producer producer = ONSFactory.createProducer(properties);
return producer;
}
}
3、Message Handler
public class MqHandler implements MessageListener {
@Override
public Action consume(Message message, ConsumeContext context) {
try {
byte[] body = message.getBody();
String content = new String(body);
// 业务代码
} catch (Exception e) {
logger.error("MessageListener.consume error:" + e.getMessage(), e);
}
return Action.CommitMessage;
}
}
4、使用 producer
@Autowired
private Producer producer;
try {
SendResult sendResult = producer.send(message);
} catch (Exception e) {
e.printStackTrace();
}
