初识ActiveMQ
博主之前的一个高并发需求:java并发(三):实例引出并发应用场景中所提到的,后来经过初步测试发现多线程并不能完全满足需求,特别是性能上的需求,或者说多线程不是比较好的解决方案,真实需求是:将商品库存(第三方数据库上)"及时"通知第三方的网购平台,达到同步商品余量信息的目的,本地是存儲了相应的阈值,在第三方数据库上的库存一旦少于库存,我们就认为这件商品已经售罄,因为要防止线上线下同一时间段销售引起的库存紧张,甚至订单已经发出但库存实际不足的情况...之前多线程定时访问库存并同步数据显然非常低效,主管老哥推荐我使用消息队列来解决问题,顿时一脸懵,消息队列是啥??
消息队列的基本概念:
消息队列(英语:message queue)是一种进程间通信或同一进程的不同线程间的通信方式,软件的贮列用来处理一系列的输入,通常是来自用户。消息队列提供了异步的通信协议,每一个贮列中的纪录包含详细说明的数据,包含发生的时间,输入设备的种类,以及特定的输入参数,也就是说:消息的发送者和接收者不需要同时与消息队列互交。消息会保存在队列中,直到接收者取回它。 ——*
博主使用的消息队列中间件是activemq,为什么用它呢?
- 多种语言和协议编写客户端。语言: java, c, c++, c#, ruby, perl, python, php。应用协议: openwire,stomp rest,wsnotification,xmpp,amqp
- 完全支持jms1.1和j2ee 1.4规范 (持久化,xa消息,事务)
- 对spring的支持,activemq可以很容易内嵌到使用spring的系统里面去,而且也支持spring2.0的特性
- 支持多种传送协议:in-vm,tcp,ssl,nio,udp,jgroups,jxta
- 支持通过jdbc和journal提供高速的消息持久化
- 从设计上保证了高性能的集群,客户端-服务器,点对点等等
以上的总结比较官方,概括来说,activemq的优势在于它是java语言开发,在基于spring的项目上容易内嵌,很大程度的减少耦合,提供可靠的任务异步处理.
activemq的通信模式:
1.点对点(queue)
- 一个消息只能被一个服务接收
- 消息一旦被消费,就会消失
- 如果没有被消费,就会一直等待,直到被消费
- 多个服务监听同一个消费空间,先到先得
2.发布/订阅模式(topic)
- 一个消息可以被多个服务接收
- 订阅一个主题的消费者,只能消费自它订阅之后发布的消息
- 消费端如果在生产端发送消息之后启动,是接收不到消息的,除非生产端对消息进行了持久化(例如广播,只有当时听到的人能听到信息)
如何实现?
业务需求是用发布-订阅模式完成,我负责消费者部分的代码,一开始是这样实现的,五步走:
- 通过连接工厂获取连接对象
- 启动连接
- 创建session
- 创建队列或topic
- 注册消息监听
public class roundrobinconfig1 { private logger logger = loggerfactory.getlogger(getclass()); @resource private inventoryservice inventoryservice; @scheduled(cron = "0 53 * * * ?")//每2分钟调度一次任务 public void operation(){ connectionfactory connectionfactory; // 连接工厂 connection connection = null; // 连接 session session; // 会话 接受或者发送消息的线程 destination destination; // 消息的目的地 messageconsumer consumer; //创建消费者 // 实例化连接工厂 connectionfactory=new activemqconnectionfactory(jmsconsumer.username, jmsconsumer.password, jmsconsumer.brokeurl); try { connection=connectionfactory.createconnection(); // 通过连接工厂获取连接 connection.start(); // 启动连接 /** * 这里的最好使用boolean.false,如果是用true则必须commit才能生效,且http://127.0.0.1:8161/admin管理页面才会更新消息队列的变化情况。 */ session=connection.createsession(boolean.false, session.auto_acknowledge); // 创建session // destination=session.createqueue("firstqueue1"); // 创建消息队列 destination=session.createtopic("firsttopic"); consumer=session.createconsumer(destination); consumer.setmessagelistener(new mylistener()); // 注册消息监听 } catch (exception e) { // todo auto-generated catch block e.printstacktrace(); } } }
具体业务逻辑写在listener里,大家使用时别忘了引入maven依赖
<!-- activemq --> <dependency> <groupid>org.apache.activemq</groupid> <artifactid>activemq-all</artifactid> <version>5.9.1</version> </dependency>
然后就进行初步测试,喜闻乐见地遇到问题了:
博主经过一通资料的查阅,依旧没有搞懂问题所在,最后问同事要来了生产者的代码,发现了问题可能出在这里:
stompjmsconnectionfactory factory = new stompjmsconnectionfactory();
生产者用了这个连接工厂获取连接,随即百度了一下stomp,了解到这其实是一种消息格式协议,另外还有amqp,openwire,mqtt等,几种消息协议的概述可以,我便换成了stompjmsconnection对象来获取连接,结果成功获取到消息体:
由于需要让订阅消息队列的程序一直运行,我采取官方推荐的死循环方式处理,并且使其在模块启动时运行,后来考虑了一下,万一死循环出现异常,那整个模块不就宕了吗,于是我给模块创建了一个子进程用来轮询消息队列,这样子进程就算挂了,整个模块也不受影响了:
import com.google.gson.gson; import com.ycyz.framework.task.domain.inventorycache; import com.ycyz.framework.task.service.inventorycacheservice; import org.fusesource.hawtbuf.buffer; import org.fusesource.stomp.jms.stompjmsconnectionfactory; import org.fusesource.stomp.jms.stompjmsdestination; import org.fusesource.stomp.jms.message.stompjmsbytesmessage; import org.slf4j.logger; import org.slf4j.loggerfactory; import org.springframework.boot.applicationarguments; import org.springframework.boot.applicationrunner; import org.springframework.core.annotation.order; import org.springframework.stereotype.component; import javax.annotation.resource; import javax.jms.*; import java.util.hashmap; import java.util.list; import java.util.map; /** * @author yhw * @classname: autorunner * @description: * @date 2019/1/7 8:27 */ @order(value = 1) @component public class autorunner implements applicationrunner { private logger logger = loggerfactory.getlogger(getclass()); map map = new hashmap(16); string result = null; string user = env("activemq_user", ""); string password = env("activemq_password", ""); string host = env("activemq_host", "域名"); string destination = "/topic/bn.stock.prod"; int port = integer.parseint(env("activemq_port", "端口号")); destination dest = new stompjmsdestination(destination); @resource private inventorycacheservice inventorycacheservice; gson gson = new gson(); stompjmsconnectionfactory factory = new stompjmsconnectionfactory(); @override public void run(applicationarguments args) throws exception{ logger.info("开始运行了..."); thread thread = new thread(){ @override public void run() { messageconsumer consumer = null; connection connection = null; long start = 0l; long count = 0l; try { factory.setbrokeruri("tcp://" + host + ":" + port); connection = factory.createconnection(user, password); connection.start(); session session = connection.createsession(false, session.auto_acknowledge); consumer = session.createconsumer(dest); start = system.currenttimemillis(); count = 1; system.out.println("waiting for messages..."); while (true) { system.out.println("轮询消息队列.."); message message = null; try { message = consumer.receive(); if (message instanceof stompjmsbytesmessage) { stompjmsbytesmessage sm = (stompjmsbytesmessage) message; buffer buffer = sm.getcontent(); byte[] a = buffer.getdata(); result = new string(a);if (result.contains("shutdown")) { long diff = system.currenttimemillis() - start; system.out.println(string.format("received %d in %.2f seconds", count, (1.0 * diff / 1000.0))); break; } //result是获取到的消息字符串,这里开始处理它 } }catch(exception e) { e.printstacktrace(); continue; } } connection.close(); }catch(jmsexception e) { e.printstacktrace(); } } }; thread.start(); } private static string env(string key, string defaultvalue) { string rc = system.getenv(key); if( rc== null ){ return defaultvalue; } return rc; } private static void flagtrigger(inventorycache inventorycache){ if(new integer(1).equals(inventorycache.getflag())){ inventorycache.setflag(0); }else{ inventorycache.setflag(1); } } private static string getresult(string thewholemessage){ int startflag = 0; int endflag = 0; for (int i = 0; i < thewholemessage.length(); i++) { if (thewholemessage.charat(i) == '{') { startflag = i; } else if (thewholemessage.charat(i) == '}') { endflag = i; } } return thewholemessage.substring(startflag, endflag + 1); } }
这样就初步完成,只要部署了多个消费者,就可以有效提升系统性能了,生产者只管往队列里"塞"待处理消息,多个消费者只管"拿"消息来处理,做到了有效的应用程序解耦.
当然我也不确定还有没有更好的方案,博主才疏学浅,懂得太少,希望有看到的大牛能够不吝赐教,谢谢了