• 1. JMS&ActiveMQ介绍By Pochmuchka
  • 2. JMS介绍 Java Message Service(JMS)是SUN提出的旨在统一各种MOM(Message-Oriented Middleware )系统接口的规范,它包含点对点(Point to Point,PTP)和发布/订阅(Publish/Subscribe,pub/sub)两种消息模型,提供可靠消息传输、事务和消息过滤等机制。 简单的说,JMS制定了一个发消息的规范。是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持。 ActiveMQ是Apache出品的开源项目,它是JMS规范的一个实现,
  • 3. JMS的作用 在不同应用之间进行通信或者从一个系统传输数据到 另外一个系统。两个应用程序之间,或分布式系统中发送消息,进行异 步通信。 这类问题有很多解决方案 ,比如DB、SOA、Socket通信、RMI,等,但我 们需要根据项目的限制以及功能和性能的需要作出选择。 JMS的应用场景:规模和复杂度较高的分布式系统。 (1)同步通信:客户发出调用后,必须等待服务对象完成处理并返回结果后才能继续执行; (2)客户和服务对象的生命周期紧密耦合:客户进程和服务对象进程都必须正常运行;如果由于服务对象崩溃或者网络故障导致客户的请求不可达,客户会接收到异常; (3)点对点通信:客户的一次调用只发送给某个单独的目标对象。
  • 4. MOM在系统中的位置
  • 5. JMS模型Java消息服务应用程序结构支持两种模型: 1.点对点模型(基于队列) 每个消息只能有一个消费者。消息的生产者和消费者之间没有时间上的 相关性.可以由多个发送者,但只能被一个消费者消费。 一个消息只能被一个接受者接受一次 生产者把消息发送到队列中(Queue),这个队列可以理解为电视机频道(channel) 在这个消息中间件上有多个这样的channel 接受者无需订阅,当接受者未接受到消息时就会处于阻塞状态 2. 发布者/订阅者模型(基于主题的) 每个消息可以有多个消费者。 生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消 费自它订阅之后发布的消息. 允许多个接受者,类似于广播的方式 生产者将消息发送到主题上(Topic) 接受者必须先订阅 注:持久化订阅者:特殊的消费者,告诉主题,我一直订阅着,即使网络断开,消息服务器也记住所有持久化订阅者,如果有新消息,也会知道必定有人回来消费。
  • 6. JMS消息发送模式
  • 7. Topic 发送模式
  • 8. JMS公共接口JMS 公共点对点域发布/订阅域ConnectionFactory QueueConnectionFactory TopicConnectionFactoryConnection QueueConnectionTopicConnectionDestination QueueTopicSession QueueSessionTopicSessionMessageProducer QueueSender TopicPublisherMessageConsumer QueueReceiverTopicSubscriber
  • 9. JMS的基本构件连接工厂: 连接工厂是客户用来创建连接的对象,例如ActiveMQ提供的 ActiveMQConnectionFactory。 连接: JMS Connection封装了JMS 客户端到JMS Provider 的连接与JMS提供者之间的一个虚拟的连接。 会话: JMS Session是生产和消费消息的一个单线程上下文。会话用于创建 消息的生产者(producer),消费者(consumer),消息(message)等, 会话,是一个事务性的上下文。 消息的生产和消费不能包含在同一个事务中。
  • 10. JMS的基本构件生产者:MessageProducer 由Session 对象创建的用来发送消息的对象 消费者:MessageConsumer 由Session 对象创建的用来发送消息的对象 消息:Message jms消息包括消息头和消息体以及其它的扩展属性。 JMS定义的消息类型有TextMessage、MapMessage、BytesMessage、 StreamMessage和ObjectMessage。 目的地:Destination,消息的目的地,是用来指定生产的消息的目标和它消费的消息的来源的对象。 消息队列:Queue 点对点的消息队列 消息主题:Tipic 发布订阅的消息队列
  • 11. Jms消息发送时序图
  • 12. Jms消息发送开发流程1、生产者(producer)开发流程(ProducerTool.java): 1.1 创建Connection: 根据url,user和password创建一个jms Connection。 1.2 创建Session: 在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标识。 1.3 创建Destination对象: 需指定其对应的主题(subject)名称,producer和consumer将根据subject来发送/接收对应的消息。 1.4 创建MessageProducer: 根据Destination创建MessageProducer对象,同时设置其持久模式。 1.5 发送消息到队列(Queue): 封装TextMessage消息,使用MessageProducer的send方法将消息发送出去。 2、消费者(consumer)开发流程(ConsumerTool.java): 2.1 实现MessageListener接口: 消费者类必须实现MessageListener接口,然后在onMessage()方法中监听消息的到达并处理。 2.2 创建Connection: 根据url,user和password创建一个jms Connection,如果是durable模式,还需要给connection设置一个clientId。 2.3 创建Session和Destination: 2.4创建replyProducer【可选】: 可以用来将消息处理结果发送给producer。 2.5 创建MessageConsumer: 根据Destination创建MessageConsumer对象。 2.6 消费message: 在onMessage()方法中接收producer发送过来的消息进行处理,并可以通过replyProducer反馈信息给producer
  • 13. Jms消息订阅者流程图
  • 14. JMS消息的事务1.创建事务createSession(paramA,paramB); paramA是设置事务的,paramB设置acknowledgment mode(应答模式) paramA设置为false时:paramB的值可为Session.AUTO_ACKNOWLEDGE,Session.CLIENT_ACKNOWLEDGE, DUPS_OK_ACKNOWLEDGE其中一个。 2.事务的应答确认 A)paramA设置为true时: paramB的值忽略, acknowledgment mode被jms服务器设置 SESSION_TRANSACTED 。 当一个事务被提交的时候,消息确认就会自动发生。 B) paramA设置为false时: Session.AUTO_ACKNOWLEDGE为自动确认,当客户成功的从receive方法返回的时候,或者从 MessageListener.onMessage方法成功返回的时候,会话自动确认客户收到的消息。 Session.CLIENT_ACKNOWLEDGE 为客户端确认。客户端接收到消息后,必须调用javax.jms.Message的 acknowledge方法。jms服务器才会删除消息。(默认是批量确认) DUPS_OK_ACKNOWLEDGE 允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会 话对象就会确认消息的接收,而且允许重复确认。如果是重复的消息,那么JMS provider必须把消息头 的JMSRedelivered字段设置为true。
  • 15. 消费者的消费方式下两种方法之一: 同步消费。通过调用消费者的receive方法从目的地中显式提取消息。receive方法可以一直阻塞到消息到达。 异步消费。客户可以为消费者注册一个消息监听器,以定义在消息到达时所采取的动作。 实现MessageListener接口,在MessageListener()方法中实现消息的处理逻辑。
  • 16. JMS 的通信机制
  • 17.     activeMQ支持多种通讯协议TCP/UDP等,我们选取最常用的TCP来分析activeMQ的通讯机制。首先我们来明确一个概念: 客户(Client):消息的生产者、消费者对activeMQ来说都叫作客户。 消息中转器(Message broker):它是activeMQ的核心,它接收信息并进行相关处理后分发给消息消费者。     为了能清楚的描述出activeMQ的核心通讯机制,我们选择3个部分来进行说明,它们分别是建立链接、关闭链接、心跳。 一、Client跟activeMQ的TCP通讯的初始化过程分析如下: 1. activeMQ初始化时,通过TcpTransportServer类根据配置打开TCP侦听端口,客户通过该端口发起建立链接的动作。 2. 把accept的Socket放入阻塞队列中。 3. 另外一个线程Socket handler阻塞着等待队列中是否有新的Socket,如果有则取出来。 4. 生成一个TransportConnection的实例。TransportConnection类的主要作用是处理链路的状态信息,并实现CommandVisitor接口来完成各类消息的处理。 5. TransportConnection会使用一个由多个TransportFilter实例组成的消息处理链条,负责对接收到的各类消息进行处理并发送相应的应答。这个链条的典型组成顺序:MutexTransport->WireFormatNegotiator->InactivityMonitor->TcpTransport。在这条链条中最后的一环就是TcpTransport类,它是实际和Client获取和发送数据的地方,该类的重要 6. 建链完成,可以进行通讯操作。方法有run()和oneway(),一个负责读取,一个负责发送。
  • 18. 二、关闭链接 activeMQ发现TCP链接的关闭,最关键的代码在TcpBufferedInputStream类中的 int n = in.read(buffer, position, buffer.length - position); 三、心跳     为了更好的维护TCP链路的使用,activeMQ采用了心跳机制作为判断双方链路的健康情况。activeMQ使用的是双向心跳,也就是activeMQ的Broker和Client双方都进行相互心跳,但不管是Broker或Client心跳的具体处理情况是完全一样的,都在InactivityMonitor类中实现,下面具体介绍。     心跳会产生两个线程“InactivityMonitor ReadCheck”和“InactivityMonitor WriteCheck”,它们都是Timer类型,都会隔一段固定时间被调用一次。ReadCheck线程主要调用的方法是readCheck(),当在等待时间内,有消息接收到,则该方法会返回true。WriteCheck线程主要调用的方法是writeCheck(),这有个小技巧,大家可以参考一下,那就是当WriteCheck线程休眠时,有任何数据发送成功,则该线程被唤醒后,不用通过TCP向对方真的发送心跳消息,这样可以从一定程度上减少网络传输的数据量。
  • 19. ActiveMQ模型分析首先介绍该模型中每个领域类的作用,然后再介绍它们之间的关系。 Broker:activeMQ的一个整体代表 RegionBroker:负责分发broker的操作到相应的消息区域 Region:activeMQ目前有四种主要消息区域:队列域(queueRegion)、主 题域(topicRegion)、临时队列域(tempQueueRegion)、临时主题域 (tempTopicRegion) TransportConnection:代表一个通讯连接 Destination:消息的目的地,主要包括两种Queue、Topic两种 Subscription:消息的消费者、订阅者 MessageStore:消息持久化存储,象比较复杂的Kaha存储机制就放在这 PendingMessageCursor:等待发给消费者的消息分发指针 ConnectionContext:用来维护发送请求所需的连接上下文
  • 20. ActiveMQ模型分析---静态模型
  • 21. ActiveMQ模型分析下面我们把这些领域类的关系进行一个描述: 1、一个RegionBroker拥有4种消息域的对象。 2、RegionBroker拥有所有目的地对象(destination)。 3、每个消息域(Region)也拥有它们对应的0或N个目的地对象(destination)。 4、同时每个Region也拥有它们对应的0或N个消息消费者、订阅者(subscription)。 5、每个目的地都有一个相应的持久化存储方式(messageStore),以及一个等待发送的消息 分发指针(pendingMessageCursor)。 6、消息消费者和目的地可以彼此拥有0或N个。 7、每个消费者都有一个对应的ConnectionContext,ConnectionContext里包括一个 TransportConnection对象,通过TransportConnection把真实的消息发给消费者。 8、TransportConnection也可以做为通讯连接,侦听消息生产者发出的信息,所以每个 TransportConnection会指向Broker对象。
  • 22. ActiveMQ模型分析-----动态模型
  • 23. ActiveMQ模型分析消费生产者进程向activeMQ所在进程发送消息和消费者消费消息的过程如上图所 示,消息传递的路径经过了核心领域模型,具体步骤如下: 步骤1:生产者通过向activeMQ为它建立好的TransportConnection发送消息给 activeMQ。 步骤2:TransportConnection对象找到RegionBroker。 步骤3:RegionBroker根据消息的类型找到对应的消息区域(Region)。 步骤4:该Region在它自己里面找到相应的消息目的地。 步骤5、6:该目的地首先根据需要进行持久化操作,并使用待发送消息指针对象。 步骤7:当有合适的消息消费者、订阅者来到时,目的地会找到这些消费者。 步骤8、9:通过该消费者对应的TransportConnection,发给相应的消费者进程。
  • 24. activeMQ消息分发指针 消息分发游标是用来保存JMS消息的引用。消息游标的处理过程如下: 1.当producer发送的持久化消息到达broker之后,broker首先会把它保存在持久存储中。 2.如果发现当前有活跃的consumer,而且这个consumer消费消息的速度能跟上producer生产 消息的速度,那么ActiveMQ会直接把消息传递给broker内部跟这个consumer关联的queue; 3 .如果当前没有活跃的consumer或者consumer消费消息的速度跟不上producer生产消息的 速度,那么ActiveMQ会使用Pending Message Cursors保存对消息的引用。 4. Pending Message Cursors把消息引用传递给broker内部跟这个consumer关联的dispatch queue。 以下是两种Pending Message Cursors: • VM Cursor。在内存中保存消息的引用。 • File Cursor。首先在内存中保存消息的引用,如果内存使用量达到上限,那么会把消息引 用保存到临时文件中。 我们可以在activemq.xml 中配置消息分发指针的存储策略。
  • 25. ActiveMQ的监控1.activeMQ自动的管理站点 http://localhost:8161/admin 2. Advisory Messages ActiveMQ 支持Advisory Messages,它允许 我们通过标准的JMS 消息来监控系统.通过它我们可以得到关 于JMS provider、producers、consumers和destinations的信息。 3. QueueBrowser 使用QueueBrowser的消息预览,编程提供监控接口。
  • 26. actviemq 配置连接URI 1.配置JMS连接最大闲置时间(消息服务器无消息) jmsBrokerURL = tcp://218.241.100.165:61616?wireFormat.maxInactivityDuration=90000 该wireFormat.maxInactivityDuration = 90000的默认值是30000ms wireFormat.maxInactivityDuration=0 这样的参数, wireFormat.maxInactivityDuration是心跳参数。 避免ActiveMQ在一段时间没有消息发送时抛出 "Channel was inactive for too long"异常。 2. maxReconnectDelay 最大重连间隔 failover:(tcp://127.0.0.1:61616?wireFormat.maxInactivityDuration=10000);maxReconnectDelay=10000 failover:(tcp://localhost:61616,tcp://remotehost:61616)?initialReconnectDelay=100 failover 失效备援 maxReconnectDelay=10000 最大重连间隔 3.设置异步发送消息 tcp://localhost:61616?jms.useAsyncSend=true tcp://localhost:61616?jms.prefetchPolicy.all=100&jms.redeliveryPolicy.maximumRedeliveries=5 4.客户端消息缓存的数量 tcp://localhost:61616?jms.prefetchPolicy.all=50 ##设置客户端最多缓存50条消息 5.客户端的预支取策略。 tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1
  • 27. ActiveMQ稳定性和容错性考虑 1.保障Jms连接 使用失效备援机制,和间隔自动重试机制,程序控制等方面来控制。 failover:(tcp://localhost:61616)?initialReconnectDelay=100&;maxReconnectAttempts=5 failover transport是一种重新连接机制,用于建立可靠的传输。此处配置的是一旦ActiveMQ broker中断,Listener端将每隔100ms自动尝试连接,直至成功连接或重试5次连接失败为止。 failover还支持多个borker同时提供服务,实现负载均衡的同时可增加系统容错性,格式: failover:(uri1,...,uriN)?transportOptions failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false failover:(uri1,...,uriN)?transportOptions failover:uri1,...,uriN failover:(tcp://localhost:61616) 2. JMSRedelivered 如果这个值为true,表示消息是被重新发送了。因为有时消费者没有确认他已经收到消息或者JMS提供者不确定消费者是否已经收到。 3.JMSExpiration 允许消息过期, setTimeToLive()设置消息的有效期。
  • 28. activeMQ的failOver重连机制"failover:(tcp://IPAddress1:61616,tcp://IPAddress1:61616)?initialReconnectDelay=100&maxReconnectAttempts=5"; 后面的参数initialReconnectDelay=100&maxReconnectAttempts=5“对每一个连接URI是通用的。 如果没有指定URI的获取方式,activeMQ会自动选择其中的一个URI来尝试建立连接(randomize 指定随机),获取连接后,ActiveMQ会维护连接的暂停和恢复。 以上面的URL为例,说明failOver的重连机制: a. IPAddress1, IPAddress2上的broker1,broker2都正常运行,创建的Connection会使用IPAddress1的broker1来发送消息,这时不激活消费者。 b.关闭broker1,Connection会自动切换到broker2的URI上来发送消息。 c. 激活消费者,消费者会先尝试broker1,由于broker1不可用,使用broker2来收消息, 这时只能收到broker2上的消息。 d.再重新启动broker1,生产者,和消费者都仍然使用broker2来发送和接受消息。 e. 关闭broker2,生产者和消费者都会自动切换到broker1上,消费者就收到之前broker发送的消息了。
  • 29. failOver重连机制
  • 30. activeMQ安全管理1.编程式实现 通过ActiveMQ提供的实现添加消息用户的权限(由SimpleAuthenticationPlugin类实现)。 2.配置实现 配置mq访问者信息, activemq安装目录下/conf/credentials.properties 权限管理 , 在${ACTIVEMQ_HOME}/conf/activemq.xml 中配置
  • 31. 调整TCP传输设置TCP传输是activeMQ最常用的传输方式。其中socketBufferSize和tcpNoDelay对传输性能有较大的影响。 socketBufferSize 通过tcp传输发送和接受数据的缓冲区大小,默认( 65536 bytes) tcpNoDelay - 默认为false。通常一个TCP socket缓冲区创建小的数据在发送之前。启用此选项 - 消息将被尽快发送。 url = "failover://(tcp://localhost:61616?tcpNoDelay=true)";
  • 32. OpenWire参数调试 parameter namedefault value descriptiontcpNoDelayEnabledFALSE消息将不会被延迟发送,每次都发送较小的数据,在网速较慢的时候可以提高性能。cacheEnabledTRUE 常用重复值(如producerId和destination)缓存使短键通过代替。这可以减少message的大小 ,在网络性能比较差的时候它对性能提高有好处。 缓存的查找增加额外的开销,会增加客户端和服务器的负载,所以添加缓存是也要基于这些综合考虑。cacheSize1024 在缓存中保存消息的最大数,该值设定不应该大于 Short.MAX_VALUE/2(即32767/2)。tightEncodingEnabledTRUE CPU密集型的方法聚集处理消息。我们会建议您关闭此功能,如果代理开始消耗所有可用的CPU:)
  • 33. wireFormat包信息程序中截获的传输格式(wire format)对象: WireFormatInfo { version=7, properties={ CacheSize=1024, CacheEnabled=true, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q] }
  • 34. ActiveMQ集群部署1.多个消息提供者 使用Network of brokers,以便在broker之间存储转发消息。 2.多个消息消费者 ActiveMQ支持订阅同一个queue的consumers上的集群。如果一个 consumer失效,那么所有未被确认(unacknowledged)的消息都 会被发送到这个queue上其它的consumers。如果某个consumer的 处理速度比其它consumers更快,那么这个consumer就会消费更多 的消息。
  • 35. ActiveMQ集群部署
  • 36. Master/salve Server1.主辅服务器的作用 主辅服务器:提供消息服务。 辅服务器:提供消息的备份,服务的备份。 2. Pure Master Slave的工作方式 A)服务端: Slave broker消费master broker上所有的消息状态,例如消息、确认和事务状态等。 Slave broker不提供消息服务。 Master broker只有在消息成功被复制到slave broker之后才会响应客户。 master broker失效的时候,slave broker可以启动network connectors和transport connectors,提供消息服 务,也可以跟着停止。 B)客户端: 使用failover的机制 uri = “failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false ”;
  • 37. Master/salve Server3.配置 Master broker不需要特殊的配置。 Slave broker需要进行以下配置: 4.限制 只能有一个slave broker连接到master broker。 master broker失效而导致slave broker成为master之后,之前的master broker只有在当前的master broker(原slave broker)停止后才能重新生效。
  • 38. spring 和activeMQ的结合使用spring对jms 的支持,配置jms的各个组件 1 配置jms连接工厂 2 配置消息队列 3 配置消息监听器 4 配置消息监听容器
  • 39. spring 和activeMQ的结合5 配置jms消息转换器 6.配置Spring的jms处理模版类(jmsTemplte) 7.消息生产者和 消息消费者 该部分可以根据业务需要,用户自己编程实现。
  • 40. ActiveMQ的消息传输机制1.The TCP Transport 语法:tcp://hostname:port?key=value 2.The Failover Transport 语法: failover:(uri1,...,uriN)?transportOptions 3.The Multicast Transport 语法: multicast://address:port?transportOptions 4.The UDP Transport 语法:udp://hostname:port?transportOptions 5. The VM Transport 语法:vm://brokerName?transportOptions 6. The NIO Transport 修改activemq.xml配置 7.The SSL Transport ssl://hostname:port?transportOptions 8. The HTTP and HTTPS Transport 语法: http://host:port https://host:port 9. The Peer Transport 语法: peer://peer-group/brokerName?brokerOptions 10.The WebSockets Transport
  • 41. 参考资料:activeMQ官网: http://activemq.apache.org/using-activemq.html 书籍:《ActiveMQ in Action .pdf》 activeMQ安全配置 http://tzj163.blog.163.com/blog/static/10961167920108763148495/ activeMQ通信机制 http://www.iteye.com/topic/426226#1090580 http://blog.163.com/xiechunhao@126/blog/static/110411383201081371758800/ http://netcomm.iteye.com/blog/421656 Activemq的内部机制 http://netcomm.iteye.com/blog/topic?page=2&show_full=true activeMQ性能: http://blog.csdn.net/yczz/article/details/6384499 http://bsnyderblog.blogspot.com/2010/01/how-to-use-automatic-failover-in.html