ActiveMQ 嵌入Tomcat

ActiveMQ 嵌入Tomcat

在一些項目中,單獨開啓一個ActiveMQ,對於項目實施來講有時略顯繁瑣。因此咱們將ActiveMQ內嵌到Tomcat,Tomcat啓動同時就順帶啓動了ActiveMQ。由此咱們須要掌握三個個重要的知識點
  • ActiveMQ中的BrokerService
  • 自啓動Servlet配置
  • 使用jconsole瞭解嵌入ActiveMQ運行狀態

1、開啓BrokerService

在pom.xml添加ActiveMQ依賴,本次代碼實例採用5.7版本,記住只須要activemq-core就行java

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-core</artifactId>
    <version>5.7.0</version>
</dependency>

在編寫BrokerService代碼部分,主要注意三個點apache

  1. 是否須要在jconsole中顯示監控信息 broker.setUseJmx(true)
  2. 設置鏈接用戶名和密碼,如何使用驗證插件
  3. 是否持久化,存儲位置設置,持久化配置

因此須要啓動一個鏈接地址 tcp://localhost:61616,用戶名爲admin,密碼爲admin,須要持久化,持久化數據文件存儲地址爲 /activemq ,須要啓動jconsole監控的BrokerService的代碼以下:session

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

2、生產者和消費者

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

  • 隊列,支持消息持久化,未消費的消息,在重啓後依然存在。如有多個消費者,在每次提取一條消息的前提下,全部消費均分隊列中的消息
  • 主題,不支持消息持久化,未消費的消息,在重啓後消息丟失。如有多個消費,每一個消費者依次消費主題中全部消息

無論是生產者仍是消費者代碼編寫,主要是4個步驟ide

  1. 創建鏈接,採用failover:()方式,自動斷線重連
  2. 創建Session,獲取發送或接收目標Destination ,指定是隊列(session.createQueue(queueName)),仍是主題(session.createTopic(topicName))
  3. 經過Session獲取生產者或消費者
  4. 生產或消費消息

咱們如今編寫一個生產者的代碼,並循環產生10條消息url

// 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 發送的消息" + i);
 System.out.println("發送消息:" + "ActiveMQ 發送的消息" + i);
 messageProducer.send(message);
}

編寫一個消費,消費上邊的10條消息spa

// 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();
        }
     }
});

運行效果插件

clipboard.png

可見,咱們生產者,產生的10條消息,已成功被消費者處理了。日誌

3、監控嵌入的ActiveMQ

對於嵌入的ActiveMQ,在BrokerService啓動前須要設置 broker.setUseJmx(true);而後找到你的JAVA_HOME,切換到bin,輸入jconsole命令。code

clipboard.png

待jconsole啓動後,選擇ActiveMQ所在的進程。鏈接後選擇Mbean頁籤

clipboard.png

紅框的地方分別爲已消費和已進入MQ中的消息的條數。選擇操做,找到那個SendTextMessage還能夠想此隊列發送消息。

4、Selvelt跟隨Tomcat啓動

對於Tomcat7.x版本以後Tomcat,Selvelt均可以經過直接在代碼中經過註解的方式配置URl鏈接,一塊兒是否自啓動loadOnStartup 這個值>=0表示須要自啓動,值越小優先級越高

// 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);
        // 這裏編寫啓動ActiveMQ代碼
 }
}

5、總結

此次以ActiveMQ做爲消息隊列使用切入點,整體上說還比較順利。其中惟一出現問題的地方就是對於activeMQ依賴過多,多依賴了jaractivemq-broker,致使消息能鏈接,但不能發送消息。後邊直接換成 activemq-all,有出現slf4j日誌衝突,使用exclusions依然不能解決問題。最終只依賴 activemq-core,完美解決全部問題。

相關文章
相關標籤/搜索