问题：
Apache Artemis客户端故障转移发现
韩阳云
我使用的是Apache Artemis V2.12.0，在两个VM中启动了两个broker实例
broker.xml(myhost1)[myhost2的broker.xml与此类似，只是我使用的端口是61616]
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<configuration xmlns="urn:activemq" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
<core xmlns="urn:activemq:core">
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/largemessages</large-messages-directory>
<paging-directory>./data/paging</paging-directory>
<!-- Connectors -->
<connectors>
<connector name="netty-connector">tcp://10.64.60.100:61617</connector><!-- direct ip addres of host myhost1 -->
<connector name="broker2-connector">tcp://myhost2:61616</connector> <!-- ip 10.64.60.101 <- mocked up ip for security reasons -->
</connectors>
<!-- Acceptors -->
<acceptors>
<acceptor name="amqp">tcp://0.0.0.0:61617?amqpIdleTimeout=0;tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP;useEpoll=true</acceptor>
</acceptors>
<cluster-connections>
<cluster-connection name="myhost1-cluster">
<connector-ref>netty-connector</connector-ref>
<retry-interval>500</retry-interval>
<use-duplicate-detection>true</use-duplicate-detection>
<message-load-balancing>ON_DEMAND</message-load-balancing>
<max-hops>1</max-hops>
<static-connectors>
<connector-ref>broker2-connector</connector-ref> <!-- defined in the connectors -->
</static-connectors>
</cluster-connection>
</cluster-connections>
<security-settings>
<security-setting match="#">
<permission type="createNonDurableQueue" roles="amq"/>
<permission type="deleteNonDurableQueue" roles="amq"/>
<permission type="createDurableQueue" roles="amq"/>
<permission type="deleteDurableQueue" roles="amq"/>
<permission type="createAddress" roles="amq"/>
<permission type="deleteAddress" roles="amq"/>
<permission type="consume" roles="amq"/>
<permission type="browse" roles="amq"/>
<permission type="send" roles="amq"/>
<permission type="manage" roles="amq"/>
</security-setting>
</security-settings>
<address-settings>
<!--default for catch all-->
<address-setting match="#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-delete-queues>false</auto-delete-queues>
<auto-delete-created-queues>false</auto-delete-created-queues>
<auto-delete-addresses>false</auto-delete-addresses>
</address-setting>
</address-settings>
</core>
</configuration>
2020-06-03 23:59:17,874 INFO [org.apache.activemq.artemis.core.server] AMQ221020: Started EPOLL Acceptor at 0.0.0.0:61617 for protocols [CORE,AMQP]
2020-06-03 23:59:17,910 INFO [org.apache.activemq.artemis.core.server] AMQ221007: Server is now live
2020-06-03 23:59:17,910 INFO [org.apache.activemq.artemis.core.server] AMQ221001: Apache ActiveMQ Artemis Message Broker version 2.12.0 [localhost, nodeID=e6c6eab6-a456-11ea-94cf-000d3a306e31]
2020-06-03 23:59:18,240 INFO [org.apache.activemq.artemis.core.server] AMQ221027: Bridge ClusterConnectionBridge@5e9820f4 [name=$.artemis.internal.sf.myhost1-cluster.bd39cc41-a201-11ea-abaa-000d3a315d06, queue=QueueImpl[name=$.artemis.internal.sf.devmq1-cluster.bd39cc41-a201-11ea-abaa-000d3a315d06, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=e6c6eab6-a456-11ea-94cf-000d3a306e31], temp=false]@2b0263f3 targetConnector=ServerLocatorImpl (identity=(Cluster-connection-bridge::ClusterConnectionBridge@5e9820f4 [name=$.artemis.internal.sf.devmq1-cluster.bd39cc41-a201-11ea-abaa-000d3a315d06, queue=QueueImpl[name=$.artemis.internal.sf.devmq1-cluster.bd39cc41-a201-11ea-abaa-000d3a315d06, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=e6c6eab6-a456-11ea-94cf-000d3a306e31], temp=false]@2b0263f3 targetConnector=ServerLocatorImpl [initialConnectors=[TransportConfiguration(name=netty-connector, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=10-64-60-100], discoveryGroupConfiguration=null]]::ClusterConnectionImpl@24293395[nodeUUID=e6c6eab6-a456-11ea-94cf-000d3a306e31, connector=TransportConfiguration(name=netty-connector, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61617&host=10-64-60-101, address=, server=ActiveMQServerImpl::serverUUID=e6c6eab6-a456-11ea-94cf-000d3a306e31])) [initialConnectors=[TransportConfiguration(name=netty-connector, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=10-64-60-100], discoveryGroupConfiguration=null]] is connected
2020-06-03 23:59:18,364 INFO [org.apache.activemq.hawtio.branding.PluginContextListener] Initialized activemq-branding plugin
步骤2:java客户机开始向代理发送消息
步骤3:从myhost1的控制台，我看到推送到队列中的消息
步骤4：停止myhost1中的代理实例
package com.demo.artemis.clients;
import java.util.Properties;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.InitialContext;
import org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory;
import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
public class ArtemisClientClustered
{
public static void main(final String[] args) throws Exception {
//only produces the message
new ArtemisClientClustered().runProducer(true, false);
}
public boolean runProducer(boolean produceMesage, boolean consumeMessage) throws Exception{
Connection connection = null;
InitialContext initalContext = null;
int i = 0;
try {
Properties jndiProp = new Properties();
jndiProp.put("java.naming.factory.initial", "org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory");
//jndiProp.put("connectionFactory.ConnectionFactory", "tcp://localhost:61616?producerMaxRate=50");
jndiProp.put("connectionFactory.ConnectionFactory", "(tcp://myhost2:61616,tcp://myhost1:61617)?ha=true;reconnectAttempts=-1;");
jndiProp.put("queue.queue/ahm.load-datawarehouse.queue","ahm.load-datawarehouse.queue");
initalContext = new InitialContext(jndiProp);
// Step 2. Perfom a lookup on the queue
Queue queue = (Queue) initalContext.lookup("queue/myExampleQ.queue");
// Step 3. Perform a lookup on the Connection Factory
//ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616?producerMaxRate=50");
ConnectionFactory cf = (ConnectionFactory)initalContext.lookup("ConnectionFactory");
// ConnectionFactory cf= new ActiveMQJMSConnectionFactory("(tcp://myhost2:61616,tcp://myhost1:61617)?ha=true;reconnectAttempts=-1;");
//using the PoolconectionFactory
JmsPoolConnectionFactory jmsPoolConnectionFactory = new JmsPoolConnectionFactory();
jmsPoolConnectionFactory.setMaxConnections(8);
jmsPoolConnectionFactory.setConnectionFactory(cf);
// Step 4. Create a JMS Connection
connection = jmsPoolConnectionFactory.createConnection("admin","admin");
// Step 5. Create a JMS Session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
if(produceMesage) {
// Step 6. Create a JMS Message Producer
MessageProducer producer = session.createProducer(queue);
System.out.println("Will now send as many messages as we can in few seconds...");
// Step 7. Send as many messages as we can in N milliseconds
final long duration = 1200000;
i=0;
long start = System.currentTimeMillis();
while (System.currentTimeMillis() - start <= duration) {
TextMessage message = session.createTextMessage("This is text message: " + i++);
producer.send(message);
}
long end = System.currentTimeMillis();
double rate = 1000 * (double) i / (end - start);
System.out.println("We sent " + i + " messages in " + (end - start) + " milliseconds");
System.out.println("Actual send rate was " + rate + " messages per second");
// Step 8. For good measure we consumer the messages we produced.
}
if(consumeMessage) {
MessageConsumer messageConsumer = session.createConsumer(queue);
connection.start();
System.out.println("Now consuming the messages...");
i = 0;
while (true) {
TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);
if (messageReceived == null) {
break;
}
i++;
}
System.out.println("Received " + i + " messages");
}
return true;
} finally {
// Step 9. Be sure to close our resources!
if (connection != null) {
connection.close();
}
}
}
}
客户端代码执行日志消息：当客户端启动时，myhost1和myhost2都在运行。过了一段时间，我手动停止myhost1代理，希望myhost2会被客户机自动发现。
....
2020-06-03 23:58:48 DEBUG ClientSessionFactoryImpl:1102 - Trying to connect with connectorFactory = org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory@45d84a20, connectorConfig=TransportConfiguration(name=ConnectionFactory, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=myhost2&reconnectAttempts=-1&ha=true
2020-06-03 23:58:48 DEBUG NettyConnector:486 - Connector NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] using native epoll
2020-06-03 23:58:48 DEBUG client:668 - AMQ211002: Started EPOLL Netty Connector version 4.1.48.Final to myhost2:61616
2020-06-03 23:58:48 DEBUG NettyConnector:815 - Remote destination: myhost2/10.64.60.101:61616
2020-06-03 23:58:48 DEBUG NettyConnector:659 - Added ActiveMQClientChannelHandler to Channel with id = cf33ff23
2020-06-03 23:58:48 DEBUG Recycler:97 - -Dio.netty.recycler.maxCapacityPerThread: 4096
2020-06-03 23:58:48 DEBUG Recycler:98 - -Dio.netty.recycler.maxSharedCapacityFactor: 2
2020-06-03 23:58:48 DEBUG Recycler:99 - -Dio.netty.recycler.linkCapacity: 16
2020-06-03 23:58:48 DEBUG Recycler:100 - -Dio.netty.recycler.ratio: 8
2020-06-03 23:58:48 DEBUG AbstractByteBuf:63 - -Dio.netty.buffer.checkAccessible: true
2020-06-03 23:58:48 DEBUG AbstractByteBuf:64 - -Dio.netty.buffer.checkBounds: true
2020-06-03 23:58:48 DEBUG ResourceLeakDetectorFactory:195 - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@6933b6c6
2020-06-03 23:58:48 DEBUG ClientSessionFactoryImpl:809 - Reconnection successful
2020-06-03 23:58:48 DEBUG NettyConnector:1269 - NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] host 1: 10.44.6.85 ip address: 10.44.6.85 host 2: myhost2 ip address: 10.44.6.85
2020-06-03 23:58:48 DEBUG ClientSessionFactoryImpl:277 - ClientSessionFactoryImpl received backup update for live/backup pair = TransportConfiguration(name=ConnectionFactory, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=myhost2&reconnectAttempts=-1&ha=true / null but it didn't belong to TransportConfiguration(name=ConnectionFactory, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=myhost2&reconnectAttempts=-1&ha=true
Will now send as many messages as we can in few seconds...
...
...
2020-06-04 00:01:09 WARN client:210 - AMQ212037: Connection failure to myhost2/10.64.60.101:61616 has been detected: AMQ219015: The connection was disconnected because of server shutdown [code=DISCONNECTED]
2020-06-04 00:01:09 DEBUG ClientSessionFactoryImpl:800 - Trying reconnection attempt 0/-1
2020-06-04 00:01:09 DEBUG ClientSessionFactoryImpl:1102 - Trying to connect with connectorFactory = org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory@45d84a20, connectorConfig=TransportConfiguration(name=ConnectionFactory, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=myhost2&reconnectAttempts=-1&ha=true
2020-06-04 00:01:09 DEBUG NettyConnector:486 - Connector NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] using native epoll
2020-06-04 00:01:09 DEBUG client:668 - AMQ211002: Started EPOLL Netty Connector version 4.1.48.Final to myhost2:61616
2020-06-04 00:01:09 DEBUG NettyConnector:815 - Remote destination: myhost2/10.64.60.101:61616
2020-06-04 00:01:09 DEBUG NettyConnector:659 - Added ActiveMQClientChannelHandler to Channel with id = d4ed884e
2020-06-04 00:01:09 DEBUG ClientSessionFactoryImpl:1063 - Connector towards NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] failed
2020-06-04 00:01:09 DEBUG ClientSessionFactoryImpl:1140 - Backup is not active, trying original connection configuration now.
2020-06-04 00:01:11 DEBUG ClientSessionFactoryImpl:800 - Trying reconnection attempt 1/-1
2020-06-04 00:01:11 DEBUG ClientSessionFactoryImpl:1102 - Trying to connect with connectorFactory = org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory@45d84a20, connectorConfig=TransportConfiguration(name=ConnectionFactory, factory=org-apache-activemq-artemis-core-remoting-impl-netty-NettyConnectorFactory) ?port=61616&host=myhost2&reconnectAttempts=-1&ha=true
2020-06-04 00:01:11 DEBUG NettyConnector:486 - Connector NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] using native epoll
2020-06-04 00:01:11 DEBUG client:668 - AMQ211002: Started EPOLL Netty Connector version 4.1.48.Final to myhost2:61616
2020-06-04 00:01:11 DEBUG NettyConnector:815 - Remote destination: myhost2/10.64.60.101:61616
2020-06-04 00:01:11 DEBUG NettyConnector:659 - Added ActiveMQClientChannelHandler to Channel with id = 1530857a
2020-06-04 00:01:11 DEBUG ClientSessionFactoryImpl:1063 - Connector towards NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] failed
020-06-04 00:01:37 DEBUG NettyConnector:659 - Added ActiveMQClientChannelHandler to Channel with id = d886a84e
2020-06-04 00:01:37 DEBUG ClientSessionFactoryImpl:1063 - Connector towards NettyConnector [host=myhost2, port=61616, httpEnabled=false, httpUpgradeEnabled=false, useServlet=false, servletPath=/messaging/ActiveMQServlet, sslEnabled=false, useNio=true] failed
2020-06-04 00:01:37 DEBUG ClientSessionFactoryImpl:1140 - Backup is not active, trying original connection configuration now.
Exception in thread "main" javax.jms.JMSException: AMQ219014: Timed out after waiting 30,000 ms for response when sending packet 71
at org.apache.activemq.artemis.core.protocol.core.impl.ChannelImpl.sendBlocking(ChannelImpl.java:457)
at org.apache.activemq.artemis.core.protocol.core.impl.ChannelImpl.sendBlocking(ChannelImpl.java:361)
at org.apache.activemq.artemis.core.protocol.core.impl.ActiveMQSessionContext.sendFullMessage(ActiveMQSessionContext.java:552)
at org.apache.activemq.artemis.core.client.impl.ClientProducerImpl.sendRegularMessage(ClientProducerImpl.java:296)
at org.apache.activemq.artemis.core.client.impl.ClientProducerImpl.doSend(ClientProducerImpl.java:268)
at org.apache.activemq.artemis.core.client.impl.ClientProducerImpl.send(ClientProducerImpl.java:143)
at org.apache.activemq.artemis.core.client.impl.ClientProducerImpl.send(ClientProducerImpl.java:125)
at org.apache.activemq.artemis.jms.client.ActiveMQMessageProducer.doSendx(ActiveMQMessageProducer.java:483)
at org.apache.activemq.artemis.jms.client.ActiveMQMessageProducer.send(ActiveMQMessageProducer.java:220)
at org.messaginghub.pooled.jms.JmsPoolMessageProducer.sendMessage(JmsPoolMessageProducer.java:182)
at org.messaginghub.pooled.jms.JmsPoolMessageProducer.send(JmsPoolMessageProducer.java:90)
at org.messaginghub.pooled.jms.JmsPoolMessageProducer.send(JmsPoolMessageProducer.java:79)
at com.demo.artemis.clients.ArtemisClientClustered.runProducer(ArtemisClientClustered.java:77)
at com.demo.artemis.clients.ArtemisClientClustered.main(ArtemisClientClustered.java:26)
Caused by: ActiveMQConnectionTimedOutException[errorType=CONNECTION_TIMEDOUT message=AMQ219014: Timed out after waiting 30,000 ms for response when sending packet 71]
... 14 more
注意：当我使用Camel消费者消费来自这个队列的消息并转换到另一个队列时。在此过程中，当我停止代理时，使用者计数会自动重定向到另一个代理实例。从控制台，我可以看到消费者计数从一个代理重定向到另一个代理。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:jee="http://www.springframework.org/schema/jee"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-3.1.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.1.xsd
http://www.springframework.org/schema/jee
http://www.springframework.org/schema/jee/spring-jee-3.1.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-3.1.xsd
http://camel.apache.org/schema/spring
http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="jmsConnectionFactory" class="org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory">
<constructor-arg index="0" value="(tcp://myhost2:61616,tcp://myhost1:61617)?ha=true;reconnectAttempts=-1;"/>
</bean>
<bean id="jmsPooledConnectionFactory" class="org.messaginghub.pooled.jms.JmsPoolConnectionFactory" init-method="start" destroy-method="stop">
<property name="maxConnections" value="10" />
<property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>
<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
<property name="connectionFactory" ref="jmsPooledConnectionFactory" />
<property name="concurrentConsumers" value="10" />
</bean>
<bean id="jms" class="org.apache.camel.component.jms.JmsComponent">
<property name="configuration" ref="jmsConfig" />
</bean>
<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
<endpoint id="queue1" uri="jms:queue:myExampleQ" />
<endpoint id="queue2" uri="jms:queue:myExampleQ2" />
<route>
<from uri="ref:queue1" />
<convertBodyTo type="java.lang.String" />
<transform>
<simple>MSG FRM queue1 TO queue2 : ${bodyAs(String)}</simple>
</transform>
<to uri="ref:queue2" />
</route>
</camelContext>
</beans>
共有1个答案
公羊新
您已经配置了一个由2个节点组成的活动/主动集群。这支持连接和消息负载平衡，但不支持透明故障转移。为了获得透明的故障转移，您需要配置主动/被动HA对。请查看ActiveMQ Artemis文档以及代理附带的HA示例，了解如何做到这一点的更多细节。
类似资料：
-
跟随wiki页面https://cwiki.apache.org/qpid/amqp-java-jms-messaging-client.html到https://cwiki.apache.org/qpid/connection-url-format.html. Simple me需要qpid-amqp-1-0-client-jms-0.20-sources。jar接受URI格式，但它似乎忽视了一
-
我使用camel和cxf组件从Web服务中获取一些数据。在某些情况下，Web服务返回标准的soap：fault。我有下一个步骤： 阅读cxf和camel邮件列表中的类似问题，我理解cxfendpoint应该抛出异常，如果有soap foalt，org.apache.cxf.binding.soap.soapfault类型的异常，但我不能得到它。onException子句的目标是处理soap错误异常
-
我们使用MQ作为传递消息的主要路径。这是我们的制度运作不可或缺的一部分。消息代理有时会失败，所有相关的队列也会随之失败。在camel中，有没有一种方法可以启动故障切换，并在其启动时恢复到主故障切换？
-
我正在尝试用6台机器实现一个Redis集群。我有一个由六台机器组成的流浪集群： 运行redis服务器 我编辑了上述所有服务器的/etc/redis/redis.conf文件，添加了这个 然后我在六台机器中的一台上运行了这个程序； Redis集群已启动并运行。我通过在一台机器上设置值手动检查它显示在其他机器上。 我的问题是，当我关闭或停止任何一台主机上的redis server时，整个集群都会停止运
-
我在一个独立的java应用程序中使用apache cxf client 3.2.2来调用一些soap Web服务。 试图在单向操作中获取soap故障代码和故障字符串，但没有成功。 我定义了一个自定义拦截器，但它永远不会到达handleMessage 定制拦截器的代码 } 我遇到的异常是MessageSenderEndingInterceptor（处于PREPARE\u SEND阶段）。 它正在抛出
-
我正在尝试将OSS Ehcache与兵马俑集群一起使用，但未能从ehcache客户端实现故障转移。 我在这里的文档中找到了这种配置 但是这个连接只接受一个URL。如果集群中有两台terracotta服务器，我希望HA具有故障转移功能。类似ActiveMQ故障转移url： 我的问题是：这在开源ehcache中是可能的，还是我需要有企业版本，如果可能的话，如何配置它。 更新 我试过<代码> 如果我尝试