摘要：嵌入在一些項(xiàng)目中，多帶帶開(kāi)啟一個(gè)，對(duì)于項(xiàng)目實(shí)施來(lái)說(shuō)有時(shí)略顯繁瑣。待啟動(dòng)后，選擇所在的進(jìn)程。連接后選擇頁(yè)簽紅框的地方分別為已消費(fèi)和已進(jìn)入中的消息的條數(shù)。

在一些項(xiàng)目中，多帶帶開(kāi)啟一個(gè)ActiveMQ，對(duì)于項(xiàng)目實(shí)施來(lái)說(shuō)有時(shí)略顯繁瑣。所以我們將ActiveMQ內(nèi)嵌到Tomcat，Tomcat啟動(dòng)同時(shí)就順帶啟動(dòng)了ActiveMQ。由此我們需要掌握三個(gè)個(gè)重要的知識(shí)點(diǎn)

ActiveMQ中的BrokerService

自啟動(dòng)Servlet配置

使用jconsole了解嵌入ActiveMQ運(yùn)行狀態(tài)

在pom.xml添加ActiveMQ依賴，本次代碼實(shí)例采用5.7版本，記住只需要activemq-core就行。

    org.apache.activemq
    activemq-core
    5.7.0

在編寫(xiě)BrokerService代碼部分，主要注意三個(gè)點(diǎn)

是否需要在jconsole中顯示監(jiān)控信息 broker.setUseJmx(true)

設(shè)置連接用戶名和密碼，如何使用驗(yàn)證插件

是否持久化，存儲(chǔ)位置設(shè)置，持久化配置

所以需要啟動(dòng)一個(gè)連接地址 tcp://localhost:61616，用戶名為admin，密碼為admin，需要持久化，持久化數(shù)據(jù)文件存儲(chǔ)地址為 /activemq ,需要啟動(dòng)jconsole監(jiān)控的BrokerService的代碼如下:

    // author:herbert qq:464884492
    BrokerService broker = new BrokerService();
    broker.setUseJmx(true); // 開(kāi)啟監(jiān)控
    broker.setPersistent(true); // 持久化
    broker.setBrokerName("Test");
    SimpleAuthenticationPlugin sap = new SimpleAuthenticationPlugin();
    AuthenticationUser au = new AuthenticationUser("admin", "admin","users");
    ArrayList d = new ArrayList();
    d.add(au);
    sap.setUsers(d); // 用戶驗(yàn)證
    broker.setPlugins(new BrokerPlugin[] { sap });
    String mqDataPath = "/activemq"; // 存儲(chǔ)位置 
    broker.getPersistenceAdapter().setDirectory(new File(mqDataPath));
    broker.addConnector("tcp://localhost:61616"); // 連接地址
    broker.start();

ActiveMQ中，通用的消息傳遞方式有兩種

隊(duì)列，支持消息持久化，未消費(fèi)的消息，在重啟后依然存在。若有多個(gè)消費(fèi)者，在每次提取一條消息的前提下，所有消費(fèi)均分隊(duì)列中的消息

主題，不支持消息持久化，未消費(fèi)的消息，在重啟后消息丟失。若有多個(gè)消費(fèi)，每個(gè)消費(fèi)者依次消費(fèi)主題中所有消息

不管是生產(chǎn)者還是消費(fèi)者代碼編寫(xiě)，主要是4個(gè)步驟

建立連接，采用failover:()方式，自動(dòng)斷線重連

建立Session，獲取發(fā)送或接收目標(biāo)Destination ，指定是隊(duì)列(session.createQueue(queueName)),還是主題(session.createTopic(topicName))

通過(guò)Session獲取生產(chǎn)者或消費(fèi)者

生產(chǎn)或消費(fèi)消息

我們現(xiàn)在編寫(xiě)一個(gè)生產(chǎn)者的代碼，并循環(huán)產(chǎn)生10條消息

 // author:herbert qq:464884492
String mqConnUrl = "tcp://localhost:61616";
String connUrl = "failover:(" + mqConnUrl.trim()+ ")?initialReconnectDelay=1000&maxReconnectDelay=30000";
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin","admin", connUrl);
javax.jms.Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("system");
MessageProducer messageProducer = session.createProducer(destination);

for (int i = 0; i < 10; i++) {
 javax.jms.TextMessage message = session.createTextMessage("ActiveMQ 發(fā)送的消息" + i);
 System.out.println("發(fā)送消息：" + "ActiveMQ 發(fā)送的消息" + i);
 messageProducer.send(message);
}

編寫(xiě)一個(gè)消費(fèi)，消費(fèi)上邊的10條消息

 // author:herbert qq:464884492
String mqConnUrl = "tcp://localhost:61616";
String connUrl = "failover:(" + mqConnUrl.trim()+ ")?initialReconnectDelay=1000&maxReconnectDelay=30000";
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", connUrl);
javax.jms.Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("system");
MessageConsumer messageConsumer = session.createConsumer(destination);
messageConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(javax.jms.Message message) {
   ActiveMQTextMessage m = (ActiveMQTextMessage) message;
    try {
          System.out.println("接收到：" + m.getText());
         } catch (JMSException e) {
         e.printStackTrace();
        }
     }
});

運(yùn)行效果

可見(jiàn)，我們生產(chǎn)者，產(chǎn)生的10條消息，已成功被消費(fèi)者處理了。

對(duì)于嵌入的ActiveMQ，在BrokerService啟動(dòng)前需要設(shè)置 broker.setUseJmx(true);然后找到你的JAVA_HOME,切換到bin，輸入jconsole命令。

待jconsole啟動(dòng)后，選擇ActiveMQ所在的進(jìn)程。連接后選擇Mbean頁(yè)簽

紅框的地方分別為已消費(fèi)和已進(jìn)入MQ中的消息的條數(shù)。選擇操作，找到那個(gè)SendTextMessage還可以想此隊(duì)列發(fā)送消息。

對(duì)于Tomcat7.x版本之后Tomcat，Selvelt都可以通過(guò)直接在代碼中通過(guò)注解的方式配置URl連接，一起是否自啟動(dòng)loadOnStartup 這個(gè)值>=0表示需要自啟動(dòng)，值越小優(yōu)先級(jí)越高

 // author:herbert qq:464884492
@WebServlet(urlPatterns = "/initmq", loadOnStartup = 1)
public class InitMqServlet extends HttpServlet {
@Override
public void init(ServletConfig config) throws ServletException {
        super.init(config);
        // 這里編寫(xiě)啟動(dòng)ActiveMQ代碼
 }
}        

這次以ActiveMQ作為消息隊(duì)列使用切入點(diǎn)，總體上說(shuō)還比較順利。其中唯一出現(xiàn)問(wèn)題的地方就是對(duì)于activeMQ依賴過(guò)多，多依賴了jaractivemq-broker，導(dǎo)致消息能連接，但不能發(fā)送消息。后邊直接換成 activemq-all,有出現(xiàn)slf4j日志沖突，使用exclusions依然不能解決問(wèn)題。最終只依賴 activemq-core,完美解決所有問(wèn)題。