消息队列：ActiveMQ：ActiveMQ的事务处理

消息队列：ActiveMQ：ActiveMQ的事务处理1消息队列基础1.1消息队列简介消息队列是一种应用程序间通信（IPC）的形式，它允许消息的发送者和接收者不需要同时存在。消息队列可以处理异步通信，使得系统组件可以独立地工作，提高系统的可扩展性和容错性。消息队列通常用于分布式系统中，以实现服务解耦、流量削峰、异步处理等功能。1.1.1消息队列的关键概念生产者(Producer):创建并发送消息到消息队列的组件。消费者(Consumer):从消息队列中接收并处理消息的组件。消息队列(MessageQueue):用于存储消息的中间件，直到它们被消费者处理。1.1.2消息队列的类型点对点(P2P):每个消息被发送到队列后，只会被一个消费者消费。发布/订阅(Pub/Sub):消息被广播到所有订阅了该主题的消费者。1.2ActiveMQ概述ActiveMQ是Apache出品的、遵循AMQP1.0协议的、功能丰富的、高性能的消息中间件。它是Apache的一个顶级项目，被广泛应用于企业级应用中，提供消息的持久化、事务处理、高可用性等功能。1.2.1ActiveMQ的特点支持多种消息协议:包括AMQP、STOMP、MQTT等。高可用性:支持主从模式、集群模式，确保消息的可靠传输。事务处理:支持本地事务和XA分布式事务，确保消息的完整性和一致性。1.3ActiveMQ的安装与配置1.3.1安装ActiveMQ下载ActiveMQ:访问ApacheActiveMQ官网，下载最新版本的ActiveMQ。解压:将下载的ActiveMQ压缩包解压到指定目录。启动:进入解压后的bin目录，运行activemq脚本启动ActiveMQ服务。#在Linux环境下启动ActiveMQ

./activemqstart1.3.2配置ActiveMQActiveMQ的配置文件位于解压目录下的conf文件夹中，主要配置文件为activemq.xml。以下是一个简单的配置示例，展示了如何设置持久化存储和监听端口。<!--activemq.xml配置示例-->

<beansxmlns="/schema/beans"

xmlns:xsi="/2001/XMLSchema-instance"

xmlns:amq="/schema/core"

xsi:schemaLocation="/schema/beans

/schema/beans/spring-beans-2.5.xsd

/schema/core

/schema/core/activemq-core.xsd">

<brokerxmlns="/schema/core"brokerName="localhost"dataDirectory="${activemq.data}">

<!--设置监听端口-->

<transportConnectors>

<transportConnectorname="openwire"uri="tcp://localhost:61616"/>

</transportConnectors>

<!--持久化存储配置-->

<persistenceAdapter>

<kahaDBdirectory="${activemq.data}/kahadb"/>

</persistenceAdapter>

</broker>

</beans>1.3.3使用Java客户端连接ActiveMQ在Java应用中，可以使用JMSAPI来连接和操作ActiveMQ。以下是一个使用ActiveMQ的Java客户端示例，展示了如何创建连接、发送和接收消息。importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.MessageConsumer;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassActiveMQExample{

publicstaticvoidmain(String[]args)throwsException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

connection.start();

//创建会话

Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

//创建队列

Destinationdestination=session.createQueue("ExampleQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//创建消息

TextMessagemessage=session.createTextMessage("Hello,ActiveMQ!");

//发送消息

producer.send(message);

//创建消息消费者

MessageConsumerconsumer=session.createConsumer(destination);

//接收消息

TextMessagereceivedMessage=(TextMessage)consumer.receive();

System.out.println("Receivedmessage:"+receivedMessage.getText());

//关闭资源

consumer.close();

producer.close();

session.close();

connection.close();

}

}在这个示例中，我们首先创建了一个连接工厂ActiveMQConnectionFactory，然后使用这个工厂创建了一个连接Connection。接着，我们创建了一个会话Session，并使用会话创建了一个队列Destination。之后，我们创建了一个消息生产者MessageProducer和一个消息消费者MessageConsumer，并通过它们发送和接收消息。最后，我们关闭了所有打开的资源，以释放系统资源。通过以上步骤，我们不仅了解了ActiveMQ的基本安装和配置，还学习了如何使用Java客户端与ActiveMQ进行交互，这对于开发基于ActiveMQ的消息驱动应用至关重要。2消息队列：ActiveMQ：ActiveMQ的事务处理2.1事务处理概念2.1.1事务处理基础在计算机科学中，事务（Transaction）是指一系列操作的集合，这些操作要么全部成功，要么全部失败，确保数据的一致性和完整性。事务处理通常遵循ACID原则：原子性（Atomicity）：事务中的所有操作要么全部完成，要么一个也不完成。一致性（Consistency）：事务开始前和结束后，数据必须保持一致状态。隔离性（Isolation）：多个事务并发执行时，它们之间不能互相干扰。持久性（Durability）：一旦事务完成，其结果将永久保存，即使系统崩溃。2.1.2消息队列中的事务在消息队列中，事务处理变得尤为重要，尤其是在处理关键业务逻辑时。例如，当一个消息的发送需要与数据库操作一起作为一个整体时，如果数据库操作成功但消息发送失败，或者消息发送成功但数据库操作失败，都会导致数据不一致。因此，消息队列中的事务处理确保了消息的发送和接收与数据库操作等其他操作一起作为一个原子操作，要么全部成功，要么全部失败。2.1.3ActiveMQ事务支持ActiveMQ，作为一款高性能的消息中间件，提供了对事务的支持，使得开发者可以在消息的发送和接收过程中使用事务，确保消息处理的可靠性和一致性。ActiveMQ的事务处理基于JTA（JavaTransactionAPI）标准，允许在消息操作中嵌入事务边界。示例：使用ActiveMQ进行事务处理importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.JMSException;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importjavax.naming.Context;

importjavax.naming.InitialContext;

importjavax.transaction.UserTransaction;

importjava.util.Properties;

publicclassActiveMQTransactionExample{

publicstaticvoidmain(String[]args){

Propertiesprops=newProperties();

props.setProperty(Context.INITIAL_CONTEXT_FACTORY,"org.apache.activemq.jndi.ActiveMQInitialContextFactory");

props.setProperty(Context.PROVIDER_URL,"tcp://localhost:61616");

try{

//创建JNDI上下文

Contextcontext=newInitialContext(props);

//查找连接工厂

ConnectionFactoryconnectionFactory=(ConnectionFactory)context.lookup("ConnectionFactory");

//查找队列

Destinationdestination=(Destination)context.lookup("dynamicQueues/MyQueue");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//开启事务支持

connection.start();

//创建会话

Sessionsession=connection.createSession(true,Session.AUTO_ACKNOWLEDGE);

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//创建事务

UserTransactionutx=(UserTransaction)context.lookup("UserTransaction");

utx.begin();

//创建并发送消息

TextMessagemessage1=session.createTextMessage("Hello,World!");

producer.send(message1);

TextMessagemessage2=session.createTextMessage("Thisisatransactionalmessage.");

producer.send(message2);

//提交事务

mit();

mit();

}catch(Exceptione){

e.printStackTrace();

}

}

}代码解释初始化JNDI上下文：通过设置Context.INITIAL_CONTEXT_FACTORY和Context.PROVIDER_URL属性，创建JNDI上下文，用于查找ActiveMQ的ConnectionFactory和Destination。查找资源：使用JNDI上下文查找ConnectionFactory和队列destination。创建连接和会话：通过ConnectionFactory创建Connection，并开启事务支持（createSession(true,Session.AUTO_ACKNOWLEDGE)）。创建消息生产者：使用会话创建MessageProducer，用于发送消息到指定的队列。事务开始：通过UserTransaction开始事务。发送消息：创建并发送两条TextMessage到队列。提交事务：使用会话的commit方法提交事务，确保消息的发送作为一个整体成功。注意事项在使用事务时，会话的创建必须指定true作为第一个参数，表示会话支持事务。事务的提交和回滚必须在事务边界内进行，否则事务将不会生效。在实际应用中，通常会使用JTA的事务管理器来管理事务，而不是直接使用UserTransaction。通过上述示例，我们可以看到ActiveMQ如何支持事务处理，确保消息的可靠发送和接收。事务处理在消息队列中是处理复杂业务逻辑的关键，它保证了即使在网络不稳定或系统故障的情况下，消息处理的一致性和完整性也能得到保障。3消息队列：ActiveMQ：ActiveMQ的事务处理3.1配置ActiveMQ事务3.1.1启用事务在ActiveMQ中，事务处理可以确保消息的完整性和一致性，尤其是在分布式系统中，事务可以保证消息的发送和接收是原子的，即要么全部成功，要么全部失败。要启用ActiveMQ的事务处理，首先需要在activemq.xml配置文件中设置<broker>元素的useJmx和plugins属性，然后在<transportConnectors>中配置transactionSupport属性。示例配置<brokerxmlns="/schema/core"brokerName="localhost"dataDirectory="${activemq.data}/broker">

<transportConnectors>

<transportConnectorname="openwire"uri="tcp://localhost:61616?maximumConnections=1000&wireFormat.maxFrameSize=10000000&transport.jmxEnabled=true&transport.transactionSupport=XA_ACKNOWLEDGE">

<!--XA_ACKNOWLEDGE表示支持事务-->

</transportConnector>

</transportConnectors>

<plugins>

<transactionIdPlugin/>

</plugins>

</broker>3.1.2配置事务持久化事务持久化是确保在系统崩溃或重启后，事务状态能够被恢复的关键。在ActiveMQ中，可以通过配置<broker>元素下的<persistenceAdapter>来实现事务的持久化。示例配置<persistenceAdapter>

<kahaDBdirectory="${activemq.data}/kahadb"/>

</persistenceAdapter>KahaDB是ActiveMQ的默认持久化存储，它能够提供高性能的事务持久化支持。3.1.3事务与消息确认在使用事务时，消息的确认机制也非常重要。在ActiveMQ中，消息确认可以通过Session.AUTO_ACKNOWLEDGE、Session.CLIENT_ACKNOWLEDGE或Session.DUPS_OK_ACKNOWLEDGE等模式进行。但在事务模式下，通常使用Session.SESSION_TRANSACTED，这意味着消息的确认是在事务提交时进行的。示例代码importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.MessageConsumer;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassJmsTransactionExample{

publicstaticvoidmain(String[]args)throwsException{

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

Connectionconnection=connectionFactory.createConnection();

connection.start();

Sessionsession=connection.createSession(true,Session.SESSION_TRANSACTED);

Destinationdestination=session.createQueue("MyQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

TextMessagemessage=session.createTextMessage("Hello,World!");

producer.send(message);

//创建消息消费者

MessageConsumerconsumer=session.createConsumer(destination);

TextMessagereceivedMessage=(TextMessage)consumer.receive();

System.out.println("Receivedmessage:"+receivedMessage.getText());

//提交事务

mit();

//关闭资源

consumer.close();

producer.close();

session.close();

connection.close();

}

}在上述代码中，createSession(true,Session.SESSION_TRANSACTED)创建了一个事务会话，这意味着所有发送和接收的消息都将在事务提交时才被确认。在消息发送和接收后，通过调用mit()来提交事务，确保消息的持久化和一致性。3.2总结通过上述配置和示例代码，我们可以看到在ActiveMQ中如何启用事务处理，配置事务持久化，以及如何在事务模式下发送和接收消息。事务处理在保证消息的可靠性和一致性方面起着至关重要的作用，尤其是在复杂的分布式系统中。4使用ActiveMQ事务4.1发送事务消息在ActiveMQ中，事务处理确保了消息的发送和接收是原子的、一致的、隔离的和持久的（ACID属性）。发送事务消息时，消息不会立即提交到队列，而是被放入一个事务上下文中，直到事务被提交。4.1.1示例代码importorg.apache.activemq.ActiveMQConnectionFactory;

importjavax.jms.*;

publicclassTransactionalProducer{

publicstaticvoidmain(String[]args){

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

try(Connectionconnection=connectionFactory.createConnection()){

connection.start();

Sessionsession=connection.createSession(true,Session.SESSION_TRANSACTED);

Destinationdestination=session.createQueue("TransactionalQueue");

MessageProducerproducer=session.createProducer(destination);

for(inti=0;i<10;i++){

TextMessagemessage=session.createTextMessage("Message"+i);

producer.send(message);

System.out.println("Sentmessage:"+message.getText());

}

mit();

System.out.println("Transactioncommitted.");

}catch(JMSExceptione){

e.printStackTrace();

}

}

}4.1.2解释创建连接工厂：使用ActiveMQConnectionFactory创建连接工厂，指定ActiveMQ的连接URL。创建连接：通过连接工厂创建Connection对象，并启动连接。创建会话：创建Session对象时，设置true表示会话是事务性的，Session.SESSION_TRANSACTED表示消息的发送和接收需要在事务中进行。创建队列：使用Session创建一个队列Destination对象。创建消息生产者：使用Session创建MessageProducer对象，用于发送消息到队列。发送消息：循环创建并发送10条TextMessage到队列，但不立即提交事务。提交事务：在所有消息发送完成后，调用mit()提交事务，确保所有消息作为一个整体被持久化。4.2接收事务消息接收事务消息时，消费者同样需要在一个事务上下文中进行操作，以确保消息的处理是原子的和持久的。4.2.1示例代码importorg.apache.activemq.ActiveMQConnectionFactory;

importjavax.jms.*;

publicclassTransactionalConsumer{

publicstaticvoidmain(String[]args){

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

try(Connectionconnection=connectionFactory.createConnection()){

connection.start();

Sessionsession=connection.createSession(true,Session.SESSION_TRANSACTED);

Destinationdestination=session.createQueue("TransactionalQueue");

MessageConsumerconsumer=session.createConsumer(destination);

for(inti=0;i<10;i++){

TextMessagemessage=(TextMessage)consumer.receive();

System.out.println("Receivedmessage:"+message.getText());

}

mit();

System.out.println("Transactioncommitted.");

}catch(JMSExceptione){

e.printStackTrace();

}

}

}4.2.2解释创建连接和会话：与发送事务消息的步骤相似，创建连接和会话，确保会话是事务性的。创建消费者：使用Session创建MessageConsumer对象，用于从队列接收消息。接收消息：循环接收10条消息，但不立即提交事务。提交事务：在所有消息接收并处理完成后，调用mit()提交事务，确保消息的接收和处理是原子的和持久的。4.3事务回滚与提交在ActiveMQ中，如果在事务处理过程中发生错误，可以回滚事务，撤销所有已发送或接收的消息。如果一切正常，可以提交事务，使消息的发送或接收生效。4.3.1示例代码importorg.apache.activemq.ActiveMQConnectionFactory;

importjavax.jms.*;

publicclassTransactionalErrorHandling{

publicstaticvoidmain(String[]args){

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

try(Connectionconnection=connectionFactory.createConnection()){

connection.start();

Sessionsession=connection.createSession(true,Session.SESSION_TRANSACTED);

Destinationdestination=session.createQueue("TransactionalQueue");

MessageProducerproducer=session.createProducer(destination);

for(inti=0;i<10;i++){

TextMessagemessage=session.createTextMessage("Message"+i);

producer.send(message);

System.out.println("Sentmessage:"+message.getText());

if(i==5){

thrownewRuntimeException("Erroroccurred,rollingbacktransaction.");

}

}

mit();

System.out.println("Transactioncommitted.");

}catch(JMSExceptione){

try{

System.out.println("Rollingbacktransactionduetoerror.");

session.rollback();

}catch(JMSExceptionrollbackException){

rollbackException.printStackTrace();

}

e.printStackTrace();

}

}

}4.3.2解释创建连接和会话：与之前的示例相同，创建连接和事务性会话。发送消息：循环发送10条消息，但在发送第6条消息时，故意抛出异常。异常处理：捕获JMSException，在异常处理中调用session.rollback()回滚事务，撤销所有已发送的消息。提交事务：如果未发生异常，调用mit()提交事务。通过以上示例，我们可以看到ActiveMQ的事务处理如何确保消息的发送和接收是可靠和一致的，即使在处理过程中发生错误，也可以通过回滚事务来撤销操作，保证系统的稳定性和数据的完整性。5事务处理最佳实践5.1确保消息一致性在ActiveMQ中，事务处理是确保消息一致性的重要机制。通过使用事务，可以保证消息的发送和接收在一个原子操作中完成，即要么全部成功，要么全部失败。这对于需要在多个系统间保持数据一致性的场景尤为重要。5.1.1实现代码示例importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassTransactionalProducer{

publicstaticvoidmain(String[]args)throwsException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话，设置为事务模式

Sessionsession=connection.createSession(true,Session.AUTO_ACKNOWLEDGE);

//创建目的地

Destinationdestination=session.createQueue("TransactionalQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//创建并发送两条消息

TextMessagemessage1=session.createTextMessage("Message1");

TextMessagemessage2=session.createTextMessage("Message2");

producer.send(message1);

producer.send(message2);

//模拟发送失败，回滚事务

//mit();//正常提交事务

session.rollback();//回滚事务

//关闭资源

session.close();

connection.close();

}

}5.1.2代码解释上述代码示例展示了如何在ActiveMQ中使用事务来发送消息。首先，我们创建了一个ActiveMQConnectionFactory实例，用于连接到本地的ActiveMQ服务器。然后，我们创建了一个Session，并将其设置为事务模式（true），这意味着我们可以通过调用commit()或rollback()方法来控制消息的发送。我们创建了两条TextMessage，并使用MessageProducer将它们发送到名为TransactionalQueue的队列中。在发送完所有消息后，我们可以通过调用mit()来提交事务，确保所有消息都被持久化。但是，如果在发送过程中发生错误，或者我们希望取消发送，可以调用session.rollback()来回滚事务，这样所有消息都不会被发送。5.2避免死信死信是指在消息队列中无法被正确处理的消息。在ActiveMQ中，可以通过设置消息的TTL（TimeToLive）和DLQ（DeadLetterQueue）来避免死信问题。5.2.1实现代码示例importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassDeadLetterQueueProducer{

publicstaticvoidmain(String[]args)throwsException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话

Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

//创建目的地

Destinationdestination=session.createQueue("NormalQueue");

//创建死信队列

Destinationdlq=session.createQueue("DeadLetterQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//创建并发送一条消息，设置TTL为10秒

TextMessagemessage=session.createTextMessage("MessagewithTTL");

message.setLongProperty("JMSExpiration",10000);

producer.send(message);

//关闭资源

session.close();

connection.close();

}

}5.2.2代码解释在本示例中，我们创建了一个消息，并设置了其TTL为10秒。这意味着如果消息在10秒内没有被消费，它将过期并被移动到死信队列（DLQ）。我们通过创建一个名为DeadLetterQueue的队列来指定DLQ。这样，即使消息在主队列中无法被正确处理，它也不会丢失，而是被转移到DLQ中，可以进行进一步的错误处理或分析。5.3优化事务性能事务处理虽然可以保证消息的一致性，但也会对性能产生影响。为了优化事务性能，可以采用批量提交的方式，即在发送多条消息后，一次性提交事务。5.3.1实现代码示例importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassBatchTransactionalProducer{

publicstaticvoidmain(String[]args)throwsException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();

//创建会话，设置为事务模式

Sessionsession=connection.createSession(true,Session.AUTO_ACKNOWLEDGE);

//创建目的地

Destinationdestination=session.createQueue("BatchQueue");

//创建消息生产者

MessageProducerproducer=session.createProducer(destination);

//创建并发送多条消息

for(inti=0;i<10;i++){

TextMessagemessage=session.createTextMessage("Message"+i);

producer.send(message);

}

//提交事务

mit();

//关闭资源

session.close();

connection.close();

}

}5.3.2代码解释在本示例中，我们创建了一个事务会话，并使用循环发送了10条消息。与单条消息发送后立即提交事务不同，我们选择在发送完所有消息后一次性提交事务。这种方式可以显著减少与服务器的交互次数，从而提高事务处理的性能。通过以上示例，我们可以看到在ActiveMQ中如何使用事务来确保消息的一致性，如何避免死信问题，以及如何优化事务性能。在实际应用中，根据业务需求合理选择和配置事务处理策略，可以有效提升消息队列的稳定性和效率。6故障排除与优化6.1事务处理常见问题在使用ActiveMQ进行事务处理时，开发者可能会遇到一些常见的问题，这些问题往往与事务的正确配置、消息的持久化以及事务的回滚机制有关。下面我们将探讨几个典型的事务处理问题，并提供相应的解决策略。6.1.1问题1：事务消息丢失原因分析：事务消息丢失通常发生在事务未正确提交或回滚的情况下。如果在事务提交前，ActiveMQ的Broker发生故障，那么未提交的事务消息可能会丢失。解决策略：-使用持久化存储：确保ActiveMQ的Broker配置了持久化存储，如KahaDB或LevelDB，这样即使Broker重启，未提交的事务消息也不会丢失。-正确配置事务：在发送事务消息时，确保在事务上下文中正确地调用mit()或session.rollback()方法。6.1.2代码示例importjavax.jms.Connection;

importjavax.jms.ConnectionFactory;

importjavax.jms.Destination;

importjavax.jms.MessageProducer;

importjavax.jms.Session;

importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassTransactionExample{

publicstaticvoidmain(String[]args)throwsException{

//创建连接工厂

ConnectionFactoryconnectionFactory=newActiveMQConnectionFactory("tcp://localhost:61616");

//创建连接

Connectionconnection=connectionFactory.createConnection();

//启动连接

connection.start();