资讯专栏INFORMATION COLUMN

ActiveMQ 嵌入Tomcat

curlyCheng / 1006人阅读

摘要:嵌入在一些项目中,多带带开启一个,对于项目实施来说有时略显繁琐。待启动后,选择所在的进程。连接后选择页签红框的地方分别为已消费和已进入中的消息的条数。

ActiveMQ 嵌入Tomcat
在一些项目中,多带带开启一个ActiveMQ,对于项目实施来说有时略显繁琐。所以我们将ActiveMQ内嵌到Tomcat,Tomcat启动同时就顺带启动了ActiveMQ。由此我们需要掌握三个个重要的知识点

ActiveMQ中的BrokerService

自启动Servlet配置

使用jconsole了解嵌入ActiveMQ运行状态

一、开启BrokerService

在pom.xml添加ActiveMQ依赖,本次代码实例采用5.7版本,记住只需要activemq-core就行


    org.apache.activemq
    activemq-core
    5.7.0

在编写BrokerService代码部分,主要注意三个点

是否需要在jconsole中显示监控信息 broker.setUseJmx(true)

设置连接用户名和密码,如何使用验证插件

是否持久化,存储位置设置,持久化配置

所以需要启动一个连接地址 tcp://localhost:61616,用户名为admin,密码为admin,需要持久化,持久化数据文件存储地址为 /activemq ,需要启动jconsole监控的BrokerService的代码如下:

    // 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 d = new ArrayList();
    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();
二、生产者和消费者

ActiveMQ中,通用的消息传递方式有两种

队列,支持消息持久化,未消费的消息,在重启后依然存在。若有多个消费者,在每次提取一条消息的前提下,所有消费均分队列中的消息

主题,不支持消息持久化,未消费的消息,在重启后消息丢失。若有多个消费,每个消费者依次消费主题中所有消息

不管是生产者还是消费者代码编写,主要是4个步骤

建立连接,采用failover:()方式,自动断线重连

建立Session,获取发送或接收目标Destination ,指定是队列(session.createQueue(queueName)),还是主题(session.createTopic(topicName))

通过Session获取生产者或消费者

生产或消费消息

我们现在编写一个生产者的代码,并循环产生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 发送的消息" + i);
 System.out.println("发送消息:" + "ActiveMQ 发送的消息" + i);
 messageProducer.send(message);
}

编写一个消费,消费上边的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();
        }
     }
});

运行效果

可见,我们生产者,产生的10条消息,已成功被消费者处理了。

三、监控嵌入的ActiveMQ

对于嵌入的ActiveMQ,在BrokerService启动前需要设置 broker.setUseJmx(true);然后找到你的JAVA_HOME,切换到bin,输入jconsole命令。

待jconsole启动后,选择ActiveMQ所在的进程。连接后选择Mbean页签

红框的地方分别为已消费和已进入MQ中的消息的条数。选择操作,找到那个SendTextMessage还可以想此队列发送消息。

四、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代码
 }
}        
五、总结

这次以ActiveMQ作为消息队列使用切入点,总体上说还比较顺利。其中唯一出现问题的地方就是对于activeMQ依赖过多,多依赖了jaractivemq-broker,导致消息能连接,但不能发送消息。后边直接换成 activemq-all,有出现slf4j日志冲突,使用exclusions依然不能解决问题。最终只依赖 activemq-core,完美解决所有问题。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/76600.html

相关文章

  • win10下安装javaee开发环境(jdk+tomcat+mysql+redis+activemq

    摘要:第步配置环境变量在中使用即可,不用写第步输入进入中,输入检验是否安装成功。安装服务启动服务停止服务切换到目录下运行参考安装解压,配置环境变量,下开启和关闭窗口方式启动安装服务启动协助服务命令校验是否启动输入参考 1.安装jdk 第1步:在官网下载和自己电脑位数相同的安装包 第2步:选择安装路径,默认在C盘,建议自定义选择其他盘符,避免c盘重装系统后,需要重新安装。 第3步:安装过程中...

    alaege 评论0 收藏0
  • Spring Boot 参考指南(构建系统)

    摘要:构建系统强烈建议你选择一个支持依赖管理的构建系统,并且可以使用发布到中心存储库的工件。例如,要升级到另一个,你可以将以下元素添加到你的在前面的示例中,我们指定了一个,但是任何依赖类型都可以以相同的方式被覆盖。 13. 构建系统 强烈建议你选择一个支持依赖管理的构建系统,并且可以使用发布到Maven中心存储库的工件。我们建议你选择Maven或Gradle,可以让Spring Boot与其...

    thursday 评论0 收藏0
  • Tomcat优化笔记

    摘要:不同的应用场景,不同的架构,不同的需求,都会对优化设置有不同要求。在这里我所记述的只是我自己在一些应用中所设置的优化项,以备不时之需,并不是放之四海而皆准的准则。为了消除这些警告,还需要根治。 一千个人眼中就有一千个哈姆雷特。——伪西方谚语 关于Tomcat的优化点之多,我估计没有上万,也有成千。不同的应用场景,不同的架构,不同的需求,都会对优化设置有不同要求。在这里我所记述的只是我...

    Coding01 评论0 收藏0
  • Spring Boot 参考指南(消息传递)

    摘要:还自动配置发送和接收消息所需的基础设施。支持是一个轻量级的可靠的可伸缩的可移植的消息代理,基于协议,使用通过协议进行通信。 32. 消息传递 Spring框架为与消息传递系统集成提供了广泛的支持,从使用JmsTemplate简化的JMS API到使用完整的基础设施异步接收消息,Spring AMQP为高级消息队列协议提供了类似的特性集。Spring Boot还为RabbitTempla...

    Doyle 评论0 收藏0

发表评论

0条评论

最新活动
阅读需要支付1元查看
<