ActiveMQ嵌入Tomcat

Java1122zzz · · 770 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

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

ActiveMQ中的BrokerService

自启动Servlet配置

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

一、开启BrokerService

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

<dependency>

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-core</artifactId>

<version>5.7.0</version>

</dependency>

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

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

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

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

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

1 // author:herbert qq:464884492

2 BrokerService broker = new BrokerService();

3 broker.setUseJmx(true); // 开启监控

4 broker.setPersistent(true); // 持久化

5 broker.setBrokerName("Test");

6 SimpleAuthenticationPlugin sap = new SimpleAuthenticationPlugin();

7 AuthenticationUser au = new AuthenticationUser("admin", "admin","users");

8 ArrayList<AuthenticationUser> d = new ArrayList<AuthenticationUser>();

9 d.add(au);

10 sap.setUsers(d); // 用户验证

11 broker.setPlugins(new BrokerPlugin[] { sap });

12 String mqDataPath = "/activemq"; // 存储位置

13 broker.getPersistenceAdapter().setDirectory(new File(mqDataPath));

14 broker.addConnector("tcp://localhost:61616"); // 连接地址

15 broker.start();

二、生产者和消费者

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

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

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

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

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

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

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

生产或消费消息

我们现在编写一个生产者的代码,并循环产生10条消息

1 // author:herbert qq:464884492

2 String mqConnUrl = "tcp://localhost:61616";

3 String connUrl = "failover:(" + mqConnUrl.trim()+ ")?initialReconnectDelay=1000&maxReconnectDelay=30000";

4 ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin","admin", connUrl);

5 javax.jms.Connection connection = connectionFactory.createConnection();

6 connection.start();

7 Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

8 Destination destination = session.createQueue("system");

9 MessageProducer messageProducer = session.createProducer(destination);

10

11 for (int i = 0; i < 10; i++) {

12 javax.jms.TextMessage message = session.createTextMessage("ActiveMQ 发送的消息" + i);

13 System.out.println("发送消息:" + "ActiveMQ 发送的消息" + i);

14 messageProducer.send(message);

15 }

编写一个消费,消费上边的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() {br/> @Override

public void onMessage(javax.jms.Message message) {

ActiveMQTextMessage m = (ActiveMQTextMessage) message;

try {

System.out.println("接收到:" + m.getText());

} catch (JMSException e) {

e.printStackTrace();

}

}

});

运行效果

ActiveMQ嵌入Tomcat

 

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

三、监控嵌入的ActiveMQ

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

ActiveMQ嵌入Tomcat

 

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

ActiveMQ嵌入Tomcat

 

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

四、Selvelt跟随Tomcat启动

对于Tomcat7.x版本之后Tomcat,Selvelt都可以通过直接在代码中通过注解的方式配置URl连接,一起是否自启动loadOnStartup这个值>=0表示需要自启动,值越小优先级越高

1 // author:herbert qq:464884492

2 @WebServlet(urlPatterns = "/initmq", loadOnStartup = 1)

3 public class InitMqServlet extends HttpServlet {

4 @Override

5 public void init(ServletConfig config) throws ServletException {

6 super.init(config);

7 // 这里编写启动ActiveMQ代码

8 }

9 }

五、总结

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

进群:可以领取免费的架构师学习资料。

进群:了解最新的学习动态

进群:了解最新的阿里,京东招聘资讯

进群:获取更多的面试资料

1、具有1-5工作经验的,面对目前流行的技术不知从何下手,需要突破技术
瓶颈的可以加群。

2、在公司待久了,过得很安逸,但跳槽时面试碰壁。需要在短时间内进修
、跳槽拿高薪的可以加群。

3、如果没有工作经验,但基础非常扎实,对java工作机制,常用设计思想
,常用java开发框架掌握熟练的,可以加群。

4、觉得自己很牛B,一般需求都能搞定。但是所学的知识点没有系统化,很
难在技术领域继续突破的可以加群。

5. 群号:835638062 点击链接加入群:https://jq.qq.com/?
_wv=1027&k=5S3kL3v
6.阿里Java高级大牛直播讲解知识点,分享知识,上面五大专题都是各位老
师多年工作经验的梳理和总结,带着大家全面、科学地建立自己的技术体系
和技术认知!


有疑问加站长微信联系(非本文作者)

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

770 次点击  ∙  1 赞  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传