前言
RocketMQ是阿里巴巴在2012年開源的分布式消息中間件殴边,目前已經(jīng)捐贈給Apache基金會憎茂,已經(jīng)于2016年11月成為 Apache 孵化項目,相信RocketMQ的未來會發(fā)揮著越來越大的作用锤岸,將有更多的開發(fā)者因此受益竖幔。
本文僅對RocketMQ的簡單實用做入門性介紹,不對RocketMQ的底層原理進行深入介紹是偷,后續(xù)文章將對RocketMQ的原理做詳細介紹拳氢。
1、RocketMQ的Maven依賴蛋铆,pom.xml中引入jar包:
<!-- RocketMQ -->
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>3.5.8</version>
</dependency>
<dependency>
<groupId>com.alibaba.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>3.5.8</version>
<type>pom</type>
</dependency>
2馋评、Spring bean 配置單例
<!-- 監(jiān)聽實現(xiàn) -->
<bean id="rocketMqListener" class="com.rocketMq.service.impl.RocketMqListenerImpl"></bean>
<!-- 監(jiān)聽配置 -->
<bean id="consumer" class="com.rocketMq.utils.Consumer" init-method="init">
<property name="consumerGroup" value="PushConsumer"/>
<property name="namesrvAddr" value="192.168.0.1:19876"/>
<property name="topic" value="node_topic"/>
<property name="subExpression" value="slu"/>
<property name="rocketMqListener" ref="rocketMqListener"/>
</bean>
3、Java源碼
1)刺啦、RocketMqListener.java
package com.rocketMq.dao;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
/**
* 消費者監(jiān)聽接口留特,業(yè)務(wù)需要實現(xiàn)此接口并配置到Consumer中
* Created by 麥克勞林
*/
public interface RocketMqListener {
boolean RocketMqMessage(MessageExt ext,ConsumeConcurrentlyContext Context);
}
2)、
package com.rocketMq.dao;
import java.util.List;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
/**
* 監(jiān)聽wrapper玛瘸,用于處理共通業(yè)務(wù)蜕青,并轉(zhuǎn)發(fā)消息到業(yè)務(wù)監(jiān)聽中。
* Created by 麥克勞林
*/
public class RocketMqWrapper implements MessageListenerConcurrently{
private RocketMqListener rocketMqListener;
public RocketMqListener getRocketMqMessageListener() {
return rocketMqListener;
}
public void setRocketMqListener(RocketMqListener rocketMqListener) {
this.rocketMqListener = rocketMqListener;
}
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messagesList,ConsumeConcurrentlyContext consumeConcurrentlyContext) {
if(ext.getTags().equals("RFID")){
if(rocketMqListener.RFIDMessage(ext, consumeConcurrentlyContext)){
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} else{
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
}
}