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,完美解决全部问题。

相关文章
相关标签/搜索