设为首页 收藏本站
查看: 1269|回复: 0

[经验分享] ActiveMQ笔记(3):基于Networks of Brokers的HA方案

[复制链接]

尚未签到

发表于 2017-3-1 06:19:50 | 显示全部楼层 |阅读模式
  上一篇介绍了基于ZK的ActiveMQ HA方案,虽然理解起来比较容易,但是有二个不足:
  1)  占用的节点数过多,1个zk集群至少3个节点,1个activemq集群也至少得3个节点,但其实正常运行时,只有一个master节点在对外响应,换句话说,花6个节点的成本只为了保证1个activemq master节点的高可用,太浪费资源了。
  2)  性能下降太明显,比起单节点的activemq,性能下降了近1个数量级。
  这一篇将介绍基于networks of brokers的HA方案,不需要借助zk等第3方组件,只需要2个activemq节点就能达到类似效果,进入正题之前,先来简单谈谈Broker这个概念。
  Broker一词的原意是『经纪人、中间人』,用在ActiveMQ的构架中,即:Broker作为Producer与Consumer的中间人(或代理人),生产者不用知道消费者在哪里、如何消费这些细节,只要将消息扔给中间人Broker即可,类似的,消费者也不用关心消息是从哪个生产者过来的,它只知道这是从Broker那里拿来的,如果画一张图来描述,就是下面这样(引用自本文最后参考文章中的图片)
DSC0000.gif

  那么,当生产者将消息发给Broker时,会发生什么?下图描述的就是这个过程:
DSC0001.gif

  1) 生产者将消息发给Broker
  2) Broker将消息落地存储
  3) 然后给生产者反馈:事情我已经办妥了!
  继续,再来看看消费者又是如何跟Broker打交道的:
DSC0002.gif

  1) Broker将接收到的消息,从db中取出
  2) 然后发送给消费者
  3) 如果消费者使用的是自动确认模式(即:Session.AUTO_ACKNOWLEDGE),则Consumer会马上告诉Broker:ok,消息我已经收到了。
  4) 然后进行自己的业务处理
  5) Broker一旦收到确认,将会马上更新消息的状态为已消费(或直接删除,取决于持久化的实现机制)(注:虽然图中步骤5排在步骤4之后,但是步骤4、5几乎是同时发生的)
  在一些大型应用中,如果一个Broker出现性能瓶颈抗不住压力,可能会拆分成多个Broker,如下图所示:
DSC0003.gif

  (注:上图中箭头的方法并非数据流向,而应该理解成调用关系,即:Producer调用Broker1,Consumer调用Broker2...)
  Producer将消息发送给Broker1,而Consumer从另一个Broker2接收消息,有点类似数据库读写分离的意思,这样系统的性能可以提升一定程度的提升,但是问题来了,Broker1上的消息,如何"同步"(见下面的注释)到Broker2呢,这就依赖networkConnector的配置。
  注:同步这个词用在这里可能不太准确,但也找不到一个更精确的词来描述,实际上,二个broker用上述机制组建成小集群后,如果生产者连接到broker1,消费者连接到broker2,当消息发送到broker1后,broker1不会将该消息复制一份到broker2,而是等消费者从broker2上消费该消息时,这条消息才从broker1取到broker2上,相当于此时broker2是消费者,从broker1消费了一条消息,然后broker2上就有这条消息了,最终消费者才能broker2上拿到这条消息。


DSC0004.gif DSC0005.gif


1 <beans
2         xmlns="http://www.springframework.org/schema/beans"
3         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
5   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
6
7   <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
8     <property name="locations">
9       <value>file:${activemq.conf}/credentials.properties</value>
10     </property>
11   </bean>
12
13   <broker xmlns="http://activemq.apache.org/schema/core" brokerName="activemq-1">
14     <networkConnectors>
15       <networkConnector uri="static:(tcp://127.0.0.1:61626)"/>
16     </networkConnectors>
17     <persistenceAdapter>
18       <kahaDB directory="${activemq.data}/kahadb"/>
19     </persistenceAdapter>
20     <transportConnectors>
21       <transportConnector name="openwire"
22                           uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
23     </transportConnectors>
24   </broker>
25
26   <import resource="jetty.xml"/>
27 </beans>
View Code  注意:14-16行及21-22行,该Broker对外暴露61616端口,同时"连接"到61626端口(即另1个broker),最终的效果相当于,如果有producer把消息发到61616(broker1),则从另一个broker(61626端口)上也能消费这条消息。
  明白这些基本原理后,在61626对应的activemq上,也做类似的配置,只不过"连接方向"正好相反,参考以下配置:





1 <beans
2         xmlns="http://www.springframework.org/schema/beans"
3         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
5   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
6
7     <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
8         <property name="locations">
9             <value>file:${activemq.conf}/credentials.properties</value>
10         </property>
11     </bean>
12
13     <broker xmlns="http://activemq.apache.org/schema/core" brokerName="activemq-2">
14         <networkConnectors>
15             <networkConnector uri="static:(tcp://127.0.0.1:61616)"/>
16         </networkConnectors>
17         <persistenceAdapter>
18             <kahaDB directory="${activemq.data}/kahadb"/>
19         </persistenceAdapter>
20         <transportConnectors>
21             <transportConnector name="openwire"
22                                 uri="tcp://0.0.0.0:61626?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
23         </transportConnectors>
24     </broker>
25
26     <import resource="jetty.xml"/>
27 </beans>
View Code  (注:如果希望2个activemq上都能访问admin管理界面,jetty.xml中的端口要修改,不要冲突)
  这样,activemq-1与activemq-2这二个broker就互为主备,发给你的消息会同步到我,发给我的消息也会同步到你,实现了HA,示意图如下:
DSC0006.png

  Producer与Consumer连接到activemq时,配置文件可以这么写:



1 <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
2     <property name="connectionFactory">
3         <bean class="org.apache.activemq.ActiveMQConnectionFactory">
4             <!--broker服务的地址-->
5             <property name="brokerURL" value="failover:(tcp://localhost:61616,tcp://localhost:61626)"/>
6             ...
7         </bean>
8     </property>
9 </bean>
  这种HA方案的优点是占用的节点数更少(只需要2个节点),而且2个broker都可以响应消息的接收与发送,性能比zookeeper方案要好一些。
  最后,再给一个简化配置的技巧,以上述的2节点HA方案中,二个activemq的配置文件都要加networkConnector配置,如果想减轻配置的工作量,也可以只在其中一个activemq设置,参考以下片段:



    <networkConnectors>
<networkConnector uri="static:(tcp://127.0.0.1:61626)" duplex="true"/>
</networkConnectors>
  即:在61616这个activemq的配置文件中,添加networkConnector时,增加duplex="true",也就是双工通讯的意思,这样61616与61626这二个activemq上的broker就建立了双向通讯连接,另一个activemq上就无需额外配置了(注:如果在61626上配置了,反而会报错)
  参考文章:
  http://www.jakubkorab.net/2011/11/understanding-activemq-broker-networks.html
  http://activemq.apache.org/networks-of-brokers.html
  最后贴二段程序运行的输出日志,以说明同步机制的正确性,打消回复中“大鹏520”的顾虑:
  背景: *.*.*.15 与 *.*.*.16 作为HA(双主)的activemq集群,发送程序只连接到15发消息,然后退出。 接收程序只从16上收消息,如果收到了,表明15上的消息同步到16。
  下面是发送程序的输出片段:(注意输出日志中关于IP的部分,这是只连接到*.*.*.15上发送的) (注:部分敏感信息,比如真实IP前缀,公司package名,用*代替了)
  14:53:09,996 <*.DemoSender> INFO [main]: 准备发送消息...
14:53:10,270 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [main]: Sending: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=9223372036854775807, Host=*.*.*.15, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]}
14:53:10,306 <org.apache.activemq.transport.InactivityMonitor> DEBUG [ActiveMQ Transport: tcp:///*.*.*.15:61616@50616]: Using min of local: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=9223372036854775807, Host=*.*.*.15, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]} and remote: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=104857600, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]}
14:53:10,306 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [ActiveMQ Transport: tcp:///*.*.*.15:61616@50616]: Received WireFormat: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=104857600, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]}
14:53:10,306 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [ActiveMQ Transport: tcp:///*.*.*.15:61616@50616]: tcp:///*.*.*.15:61616@50616 before negotiation: OpenWireFormat{version=11, cacheEnabled=false, stackTraceEnabled=false, tightEncodingEnabled=false, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
14:53:10,307 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [ActiveMQ Transport: tcp:///*.*.*.15:61616@50616]: tcp:///*.*.*.15:61616@50616 after negotiation: OpenWireFormat{version=11, cacheEnabled=true, stackTraceEnabled=true, tightEncodingEnabled=true, sizePrefixDisabled=false, maxFrameSize=104857600}
14:53:10,416 <org.springframework.jms.core.JmsTemplate> DEBUG [main]: Executing callback on JMS Session: PooledSession { ActiveMQSession {id=ID:yangjunmings-MacBook-Pro.local-50615-1461480790088-1:1:1,started=false} java.lang.Object@5a56cdac }
14:53:10,471 <org.springframework.jms.core.JmsTemplate> DEBUG [main]: Sending created message: ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = null, originalDestination = null, originalTransactionId = null, producerId = null, destination = null, transactionId = null, expiration = 0, timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = false, type = null, priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = message test:0}
14:53:10,516 <*.DemoSender>  INFO [main]: 消息0发送完成!
14:53:11,520 <org.springframework.jms.core.JmsTemplate> DEBUG [main]: Executing callback on JMS Session: PooledSession { ActiveMQSession {id=ID:yangjunmings-MacBook-Pro.local-50615-1461480790088-1:1:1,started=false} java.lang.Object@5a56cdac }
14:53:11,521 <org.springframework.jms.core.JmsTemplate> DEBUG [main]: Sending created message: ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = null, originalDestination = null, originalTransactionId = null, producerId = null, destination = null, transactionId = null, expiration = 0, timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = false, type = null, priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = message test:1}
  运行完以后,关掉发送程序,然后启动接收程序
  14:54:40,965 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [main]: Sending: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=9223372036854775807, Host=*.*.*.16, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]}
14:54:41,002 <org.apache.activemq.transport.InactivityMonitor> DEBUG [ActiveMQ Transport: tcp:///*.*.*.16:61616@50642]: Using min of local: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=9223372036854775807, Host=*.*.*.16, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]} and remote: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=104857600, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]}
14:54:41,003 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [ActiveMQ Transport: tcp:///*.*.*.16:61616@50642]: Received WireFormat: WireFormatInfo { version=11, properties={TcpNoDelayEnabled=true, SizePrefixDisabled=false, CacheSize=1024, StackTraceEnabled=true, CacheEnabled=true, TightEncodingEnabled=true, MaxFrameSize=104857600, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000}, magic=[A,c,t,i,v,e,M,Q]}
14:54:41,003 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [ActiveMQ Transport: tcp:///*.*.*.16:61616@50642]: tcp:///*.*.*.16:61616@50642 before negotiation: OpenWireFormat{version=11, cacheEnabled=false, stackTraceEnabled=false, tightEncodingEnabled=false, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
14:54:41,004 <org.apache.activemq.transport.WireFormatNegotiator> DEBUG [ActiveMQ Transport: tcp:///*.*.*.16:61616@50642]: tcp:///*.*.*.16:61616@50642 after negotiation: OpenWireFormat{version=11, cacheEnabled=true, stackTraceEnabled=true, tightEncodingEnabled=true, sizePrefixDisabled=false, maxFrameSize=104857600}
14:54:41,163 <org.apache.activemq.thread.TaskRunnerFactory> DEBUG [main]: Initialized TaskRunnerFactory[ActiveMQ Session Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@485966cc[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
14:54:41,181 <*.DemoListener>  INFO [ActiveMQ Session Task-1]: message test:0
  
注意接收程序中IP的部分,这是从*.*.*.16上接收到的,说明消息已经从 *.*.*.15同步到*.*.*.16上了,否则不可能收到消息。

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.yunweiku.com/thread-348470-1-1.html 上篇帖子: Spark安装过程 下篇帖子: Solr学习之三 solr配置说明之一
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表