從上一篇開始說起
上一篇從零開始學 Java - Spring 集成 ActiveMQ 配置（一）文章中講了我關於消息隊列的思考過程，現在這一篇會講到 ActivMQ 與 Spring 框架的整合配置，以及我們的多隊列配置、斷線重連機制，當然還有的生產者及消費監聽者的具體代碼實現。
ActiveMQ 是 Apache 出品，最流行的、能力強勁的開源消息總線。ActiveMQ 是一個完全支持 JMS1.1 和 J2EE 1.4 規范的 JMS Provider 實現，可以很容易內嵌到使用Spring的系統里面去，所以我們選擇它。
一些基礎的准備工作
先去官網下載服務器端的安裝包，地址：http://activemq.apache.org/activemq-5130-release.html， 根據自己系統下來，並進行安裝，這里不再贅述，安裝完成后，一般可以打開一個這樣的界面：
接着，找到我們需要的 jar 包。其實剛剛下載的文件中就有，我們只需要導入到我們的項目中即可，當然，你也可以從我的示例項目中獲得：https://github.com/mafly/SpringDemo/blob/master/WebContent/WEB-INF/lib/
開始寫代碼吧
一、建立cn.mayongfa.activemq包。
1.新建MessageSender.java消息發送類
/**
* ActiveMQ 消息生產類
*
* @author Mafly
*
*/
@Component
public class MessageSender {
private Logger log = Logger.getLogger(MessageSender.class);
@Autowired
private JmsTemplate jmsTemplate;
private String Queue = "default_queue";
private String GoldQueue = "gold_queue";
private Gson gson = new Gson();
/**
* 用戶登錄消息
*/
public void userLogin(long id, String username) {
Map<String, Object> map = new HashMap<String, Object>();
map.put("userid", id);
map.put("username", username);
System.out.println("發送了一條消息。");
// 發送到金幣隊列
sendMessage(gson.toJson(map), 1);
}
/**
* 發送到消息隊列
*
* @param messgae
* @param type
* 類型，0:默認隊列 1：金幣隊列 ...
*/
public void sendMessage(final String messgae, int type) {
try {
String destination = this.Queue;
if (type == 1) {
destination = GoldQueue;
}
jmsTemplate.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage(messgae);
return textMessage;
}
});
} catch (Exception e) {
log.error("", e);
}
}
}
這個類是用來往消息隊列發送信息的。
2.新建ConsumerMessageListener.java消息監聽類，繼承於MessageListener
/**
* 消費者監聽類
*
* @author Mafly
*/
@Component
public class ConsumerMessageListener implements MessageListener {
private Logger log = Logger.getLogger(ConsumerMessageListener.class);
@Override
public void onMessage(Message arg0) {
// 監聽發送到消息隊列的文本消息，作強制轉換。
TextMessage textMessage = (TextMessage) arg0;
try {
System.out.println("接收到的消息內容是：" + textMessage.getText());
// TODO: 你喜歡的任何事情...
} catch (JMSException e) {
log.error("", e);
}
}
}
這個類就是具體的消息消費類，我們就可以在這實現具體的業務邏輯處理了。
3.新建ActiveMQTransportListener.java消息傳輸監聽類，繼承於TransportListener
/**
* 消息傳輸監聽
* @author Mafly
*
*/
public class ActiveMQTransportListener implements TransportListener {
private Logger log = Logger.getLogger(ActiveMQTransportListener.class);
/**
* 對消息傳輸命令進行監控
* @param command
*/
@Override
public void onCommand(Object o) {
}
/**
* 對監控到的異常進行觸發
* @param error
*/
@Override
public void onException(IOException error) {
log.error("onException -> 消息服務器連接錯誤......", error);
}
/**
* 當failover時觸發
*/
@Override
public void transportInterupted() {
log.warn("transportInterupted -> 消息服務器連接發生中斷...");
//這里就可以狀態進行標識了
}
/**
* 監控到failover恢復后進行觸發
*/
@Override
public void transportResumed() {
log.info("transportResumed -> 消息服務器連接已恢復...");
//這里就可以進行狀態標識了
}
}
這里就可以做消息傳輸過程中的一些監控及處理，比如常見的異常處理等等。
二、在applicationContext.xml文件中配置
重要的是建完包之后要在applicationContext.xml文件中配置掃描包，完成 Bean 的注入 。就是下面：
<context:component-scan base-package="cn.mayongfa.activemq" />
在applicationContext.xml文件中加入 JMS 的配置：
<!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<!-- 真正可以產生Connection的ConnectionFactory，由ActiveMQ提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="failover:tcp://127.0.0.1:61616" />
<!-- 消息傳輸監聽器 處理網絡及服務器異常 -->
<property name="transportListener">
<bean class="cn.mayongfa.activemq.ActiveMQTransportListener" />
</property>
</bean>
<!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory" />
</bean>
<!--這個是隊列目的地 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>default_queue,gold_queue</value>
</constructor-arg>
</bean>
<!-- 消息監聽器 -->
<bean id="consumerMessageListener" class="cn.mayongfa.activemq.ConsumerMessageListener" />
<!-- 消息監聽容器 -->
<bean id="jmsContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
<!-- 設置固定的線程數 -->
<property name="concurrentConsumers" value="2"></property>
<!-- 設置動態的線程數 -->
<property name="concurrency" value="2-5"></property>
</bean>
這里的消息服務器地址及隊列名稱也可以像緩存一樣放在配置文件。當然，你不能忘了把配置文件讀取到 Spring 容器中管理。
- queueDestination：這個節點下面的
default_queue,gold_queue這就是兩個隊列的配置了，多隊列在這里用逗號分隔配置就可以了。 - failover：從字面意思你應該能明白這個意思，就是啟用斷線重連機制的。我們也是可以在
ActiveMQTransportListener類里邊監聽到的。
配置完成，代碼也寫完了，接下來就是調用測試了。
總結一下
這篇文章有了干貨，用具體的代碼及配置完成了簡單的消息隊列的發送、監聽等等，還有多隊列的一些處理，以及斷線重連機制。希望我的分享對你有所幫助吧，所有這些配置及代碼都可以在我的 GitHub 上關於 Spring 的示例項目看到：https://github.com/mafly/SpringDemo/tree/activemq。