zoukankan      html  css  js  c++  java
  • Apache ActiveMQ实战(1)-基本安装配置与消息类型

    ActiveMQ简介




    ActiveMQ是一种开源的,实现了JMS1.1规范的,面向消息(MOM)的中间件,为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。ActiveMQ使用Apache提供的授权,任何人都可以对其实现代码进行修改。


    ActiveMQ的设计目标是提供标准的,面向消息的,能够跨越多语言和多系统的应用集成消息通信中间件。ActiveMQ实现了JMS标准并提供了很多附加的特性。这些附加的特性包括,JMX管理(java Management Extensions,即java管理扩展),主从管理(master/salve,这是集群模式的一种,主要体现在可靠性方面,当主中介(代理)出现故障,那么从代理会替代主代理的位置,不至于使消息系统瘫痪)、消息组通信(同一组的消息,仅会提交给一个客户进行处理)、有序消息管理(确保消息能够按照发送的次序被接受者接收)。


    ActiveMQ 支持JMS规范,ActiveMQ完全实现了JMS1.1规范。


    JMS规范提供了同步消息和异步消息投递方式、有且仅有一次投递语义(指消息的接收者对一条消息必须接收到一次,并且仅有一次)、订阅消息持久接收等。如果仅使用JMS规范,表明无论您使用的是哪家厂商的消息代理,都不会影响到您的程序。


    ActiveMQ整体架构


    ActiveMQ主要涉及到5个方面:


    • 传输协议
    消息之间的传递,无疑需要协议进行沟通,启动一个ActiveMQ打开了一个监听端口, ActiveMQ提供了广泛的连接模式,其中主要包括SSL、STOMP、XMPP;ActiveMQ默认的使用 的协议是openWire,端口号:61616;

    • 消息域
    ActiveMQ主要包含Point-to-Point (点对点),Publish/Subscribe Model (发布/订阅者),其中在 Publich/Subscribe 模式下又有Nondurable subscription和       durable subscription (持久 化订阅)2种消息处理方式

    • 消息存储
    在消息传递过程中,部分重要的消息可能需要存储到数据库或文件系统中,当中介崩溃时,信息不 回丢失

    • Cluster  (集群)
    最常见到 集群方式包括network of brokers和Master Slave;

    • Monitor (监控)
    ActiveMQ一般由jmx来进行监控;




    ActiveMQ的安装配置


    1. 通过http://activemq.apache.org/download.html 下载:apache-activemq-5.13.3-bin.tar.gz 
    2. 把下载的该文件通过tar –zxvf apache-activemq-5.13.3-bin.tar.gz解压在当前目录
    3. 通过修改$ACTIVEMQ_HOME/conf/activemq.xml文件可以修改其配置
    一般修改的其实也只有以下几个段:、


     <destinationPolicy>
                <policyMap>
                  <policyEntries>
    

    我们在此段増加配置如下:


    <destinationPolicy>
                <policyMap>
                  <policyEntries>
                    <policyEntry topic=">" >
                    <pendingMessageLimitStrategy>
                        <constantPendingMessageLimitStrategy limit="50000"/>
                      </pendingMessageLimitStrategy>
                   </policyEntry>
                   <policyEntry queue=">" producerFlowControl="false" optimizedDispatch="true" memoryLimit=“2mb">
                   </policyEntry>
                  </policyEntries>
                </policyMap>
    </destinationPolicy>
    

    此处,我们使用的是”>”通配符,上述配置为每个队列、每个Topic配置了一个最大2mb的队列,并且使用了”optimizedDispatch=true”这个策略,该策略会启用优化了的消息分发器,直接减少消息来回时的上下文以加快消息分发速度。


    找到下面这一段


     <persistenceAdapter>
                <kahaDB directory="${activemq.data}/kahadb"/>
     </persistenceAdapter>
    

    为确保扩展配置既可以处理大量连接也可以处理海量消息队列,我们可以使用JDBC或更新更快的KahaDB消息存储。默认情况下ActiveMQ使用KahaDB消息存储。


    ActiveMQ支持三种持久化策略:AMQ、KAHADB、JDBC


    • AMQ 它是一种文件存储形式,具有写入快、容易恢复的特点,采用这种方式持久化消息会被存储在一个个文件中,每个文件默认大小为32MB,如果一条消息超过32MB,那么这个值就需要设大。当一个文件中所有的消息被“消费”掉了,那么这文件会被置成“删除”标志,并且在下一个清除开始时被删除掉。


    • KAHADB,相比AMQ来説,KAHADB速度没有AMQ快,可是KAHADB具有极强的垂直和横向扩展能力,恢复时间比AMQ还要短,因此从5.4版后ActiveMQ默认使用KAHADB作为其持久化存储。而且在作MQ的集群时使用KAHADB可以做到Cluster+Master Slave的这样的完美高可用集群方案。


    • JDBC,即ActiveMQ默认可以支持把数据持久化到DB中,如:MYSQL、ORACLE等。

    找到下面这一段

    <systemUsage>
                <systemUsage>
                    <memoryUsage>
                        <memoryUsage percentOfJvmHeap="90" />
                    </memoryUsage>
                    <storeUsage>
                        <storeUsage limit="100 gb"/>
                    </storeUsage>
                    <tempUsage>
                        <tempUsage limit="50 gb"/>
                    </tempUsage>
                </systemUsage>
    </systemUsage>
    

    此处为ActiveMQ的内存配置,从5.10版后ActiveMQ在<memoryUsage>中引入了一个percentOfJvmHeap的配置,该百分比为:


    $ACTIVEMQ_HOME/bin/env中配置的JVM堆大小的百分比,如$ACTIVEMQ_HOME/bin/env 中:

    # Set jvm memory configuration (minimal/maximum amount of memory)
    ACTIVEMQ_OPTS_MEMORY="-Xms2048M -Xmx2048M"
    

    那么此处的percentOfJvmHeap=90即表示:MQ消息队列一共会用到2048M*0.9的内存。

    全部配完后我们可以通过以下命令启动ActiveMQ

    $cd $ACTIVEMQ_HOME
    $ ./activemq console  
    

    这种方式为前台启动activemq,用于开发模式便于调试时的信息输出。


    你也可以使用:

    以后台进程的方式启动activemq。


    启动后在浏览器内输入http://192.168.0.101:8161/admin/ 输入管理员帐号(默认为admin/admin)即可登录activemq的console界面



    • 启动后的ActiveMQ的数据位于:$ACTIVEMQ_HOME/data/目录内


    • 启动后的ActiveMQ运行日志位于:$ACTIVEMQ_HOME/data/目录内的activemq.log文件


    • 如果需要改ActiveMQ的日志配置可以通过修改$ACTIVEMQ_HOME/conf/log4j.properties



    ActiveMQ与Spring集成



    在Spring中建立一个activemq.xml文件,使其内容如下:

    <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd">
    	<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供 -->
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    
    		<property name="brokerURL" value="tcp://192.168.0.101:61616" />
    		<property name="useAsyncSend" value="true" />
    		<property name="alwaysSessionAsync" value="true" />
    		<property name="useDedicatedTaskRunner" value="true" />
    	</bean>
    	<!-- 发送消息的目的地(一个队列) -->
    	<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
    		<!-- 设置消息队列的名字 -->
    		<constructor-arg value="ymk.queue?consumer.prefetchSize=100" />
    	</bean>
    </beans>
    


    其中:


    <property name="alwaysSessionAsync" value=“true" />


    对于一个connection如果只有一个session,该值有效,否则该值无效,默认这个参数的值为true。


    <property name="useAsyncSend" value="true" />


    将该值开启官方说法是可以取得更高的发送速度(5倍)。


    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
    <!-- 设置消息队列的名字 -->
    <constructor-arg value="ymk.queue?consumer.prefetchSize=100" />
    </bean>


    在此我们申明了一个队列,并用它用于后面的实验代码。


    consumer.prefetchSize则代表我们在此使用“消费者”预分配协议,在消费者内在足够时可以使这个值更大以获得更好的吞吐性能。


    工程中的pom.xml文件主要内容如下:

    。。。。。。
    <properties>
    	<activemq_version>5.13.3</activemq_version>
    </properties>
    。。。。。。
    <dependency>
    	<groupId>org.apache.activemq</groupId>
    	<artifactId>activemq-all</artifactId>
    	<version>${activemq_version}</version>
    </dependency>
    
    <dependency>
    	<groupId>org.apache.activemq</groupId>
    	<artifactId>activemq-pool</artifactId>
    	<version>${activemq_version}</version>
    </dependency>
    


    ActiveMQ与Spring集成-发送端代码


    package webpoc;
    
    public class AMQSender {
    	public static void sendWithAuto(ApplicationContext context) {
    		ActiveMQConnectionFactory factory = null;
    		Connection conn = null;
    		Destination destination = null;
    		Session session = null;
    		MessageProducer producer = null;
    		try {
    			destination = (Destination) context.getBean("destination");
    			factory = (ActiveMQConnectionFactory) context.getBean("targetConnectionFactory");
    			conn = factory.createConnection();
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			producer = session.createProducer(destination);
    			Message message = session.createTextMessage("...Hello JMS!");
    			producer.send(message);
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			try {
    				producer.close();
    				producer = null;
    			} catch (Exception e) {
    			}
    			try {
    				session.close();
    				session = null;
    			}
    			} catch (Exception e) {
    			}
    			try {
    				conn.stop();
    			} catch (Exception e) {
    			}
    			try {
    				conn.close();
    			} catch (Exception e) {
    			}
    		}
    	}
    	public static void main(String[] args) {
    		final ApplicationContext context = new ClassPathXmlApplicationContext("classpath:/spring/activemq.xml");
    		sendWithAuto(context);
    	}
    
    }
    


    ActiveMQ与Spring集成-接收端代码


    package webpoc;
    
    public class TranQConsumer extends Thread implements MessageListener {
    
    	private Connection conn = null;
    	private Destination destination = null;
    	private Session session = null;
    	public void run() {
    		receive();
    	}
    	public void receive() {
    		ConnectionFactory factory = null;
    		Connection conn = null;
    		try {
    			final ApplicationContext context = new ClassPathXmlApplicationContext("classpath:/spring/activemq.xml");
    			factory = (ActiveMQConnectionFactory) context.getBean("targetConnectionFactory");
    			conn = factory.createConnection();
    			conn.start();
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			destination = (Destination) context.getBean("destination");
    			MessageConsumer consumer = session.createConsumer(destination);
    			consumer.setMessageListener(this);
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    	public void onMessage(Message message) {
    
    		try {
    			TextMessage tm = (TextMessage) message;
    			System.out.println("TranQConsumer receive message: " + tm.getText());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    	public static void main(String[] args) {
    		TranQConsumer tranConsumer = new TranQConsumer();
    		tranConsumer.start();
    	}
    }
    


    ActiveMQ与Spring集成-示例讲解


    上述例子非常的简单。


    它其实是启动了一个Message Listener用来监听ymk.queue中的消息,如果有消息到达,接收端代码就会把消息“消费”掉。


    而发送端代码也很简单,它每次向ymk.queue队列发送一个文本消息。


    这边所谓的MQ消费大家可以这样理解:


    用户sender向MQ的KAHADB中插入一条数据。


    用户receiver把这条数据select后,再delete,这个select一下后再delete就是一个“消费”动作。


    简单消息与事务型消息


    我们可以注意到上述的例子中我们的代码中有这样的一段: 

    session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    

    它代表的是我们的MQ消费端消费模式为“自动”,即一旦消费端从MQ中取到一条消息,这条消息会自动从队列中删除。


    ActiveMQ是一个分布式消息队列,它自然支持“事务”型消息,我们可以举一个例子


    系统A和系统B是有一个事务的系统间“服务集成”,我们可以把它想成如下场景:


    系统A先会do sth…然后发送消息给系统B,系统B拿到消息后do sth,如果在其中任意一个环节发生了Exception,那么代表系统A与系统B之间的消息调用这一过程为“失败”。


    失败要重发,重发的话那原来那条消息必须还能重新拿得到。


    此时我们就需要使用事务性的消息了。而事务性的消息是在:


    生产端和消费端在创建session时,需要:


    session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);


    下面来看一个实际例子。


    事务型消息发送端(生产端)


    此处其它代码与普通式消息发送代码相似,只在以下几处有不同,首先在取得session时会声明事务开启“true”。

    session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
    

    然后在发送时会有一个动作:

    producer.send(message);
    System.out.println("send......" + Thread.currentThread().getId());
    session.commit();
    

    相应的在catch(Exception)时需要

    catch (Exception e) {
    						e.printStackTrace();
    	try {
    		session.rollback();
    	} catch (Exception ex) {
    	}
    } 
    


    事务型消息接收端(消费端)


    在我们的接收端的createSession时也需要把它设为“事务开启”,此时请注意,生产和消费是在一个事务边界中的。

    session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
    

    然后在接收时会有一个动作:

    try {
    	TextMessage tm = (TextMessage) message;
    	System.out.println("TranQConsumer receive message: " + tm.getText());
    	session.commit();
    
    } catch (Exception e) {
    	e.printStackTrace();
    	try {
    		session.rollback();
    	} catch (Exception ex) {
    	}
    }
    

    注意:


    1. 如果在消费端的onMessage中没有session.commit(),那么这条消息可以正常被接收,但不会被消费,换句话説客户端只要不commit这条消息,这条消息可以被客户端无限消费下去,直到commit(从MQ所persistent的DB中被删除)。
    2. 如果在消费断遇到任何Exception时session.rollback()了,ActiveMQ会按照默认策略每隔1s会重发一次,重发6次如果还是失败,则进入ActiveMQ的ActiveMQ.DLQ队列,重发策略这个值可以设(稍后会给出)。
    3. 如果在生产端的try{}块里发生错误,导致回滚(没有commit),会怎么样?消费队列永远拿不到这条被rollback的消息,因为这条数据还没被插入KAHADB中呢。
    4. 再如果,消费端拿到了消息不commit也不rollback呢?那消费端重启后会再次拿到这条消息(因为始终取where status=‘未消费’取不到的原因,对吧?)

    事务型消息的重发机制

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    
    	<property name="brokerURL" value="tcp://192.168.0.101:61616" />
    	<property name="useAsyncSend" value="true" />
    	<property name="alwaysSessionAsync" value="true" />
    	<property name="useDedicatedTaskRunner" value="true" />
    	<property name="redeliveryPolicy" ref="activeMQRedeliveryPolicy" />
    </bean>
    <!-- 发送消息的目的地(一个队列) -->
    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
    	<!-- 设置消息队列的名字 -->
    	<constructor-arg value="ymk.queue?consumer.prefetchSize=100" />
    </bean>
    
    <amq:redeliveryPolicy id="activeMQRedeliveryPolicy"
    	destination="#destination" redeliveryDelay="100"maximumRedeliveries="1" />
    


    以上例子申明了对于destination这个队列的重发机制为间隔100毫秒重发一次。


    事务型消息的演示








    点对点,应答式消息


    所谓点对点应答式消息和事务无关,它主要实现的是如:


    生产端:我发给你一个消息了,在你收到并处理后请回复!因为我要根据你的回复内容再做处理


    消费端:我收到你的消息了,我处理完了请查收我给你的回复


    生产端:收到你的消息,88




    点对点,应答式消息核心代码-配置部分


    <!-- 发送消息的目的地(一个队列) -->
    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
    <!-- 设置消息队列的名字 -->
    <constructor-arg value="ymk.queue?consumer.prefetchSize=100" />
    </bean>
    
    <bean id="replyDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <!-- 设置消息队列的名字 -->
    <constructor-arg value="ymk.reply.queue" />
    </bean>
    



    其实也没啥花头,就是多了一个队列(不要打我)


    。。。。。。


    关键在于代码,代码,不要只重视表面吗。。。要看内含的LA。。。


    这两个队列其实:


    一个Request
    一个应答(也可以使用temp队列来做应答队列)


    点对点,应答式消息核心代码-设计部分


    我们设立两个程序:


    • 发送端(生产端)内含一个MessageListener,用来收消费端的返回消息
    • 服务端(消费端)内含一个MessageListener,用来收生产端发过来的消息然后再异步返回


    而沟通生产端和消费端的这根“消息链”是两个东西:


    • JMSCorrelationID
    • JMSReplyTo


    JMSCorrelationID:
    它就是一个随机不可重复的数字,以String型传入API,也可以是GUID,它主要是被用来标示MQ 中每一条不同的消息用的一个唯一ID


    JMSReplyTo
    它就是一个生产端用来接收消费端返回消息的地址


    点对点,应答式消息核心代码-生产端部分代码


    String correlationId = RandomStringUtils.randomNumeric(5);
    
    consumer = session.createConsumer(replyDest);
    message.setJMSReplyTo(replyDest);
    message.setJMSCorrelationID(correlationId);
    consumer.setMessageListener(this);
    


    • RandomStringUtils
    import org.apache.commons.lang.RandomStringUtils;


    • replyDest
    replyDest = (Destination) context.getBean("replyDestination");


    来看位于客户端(生产端)的messageListener吧

    public void onMessage(Message message) {
    	TextMessage tm = (TextMessage) message;
    	try {
    		System.out.println("Client接收Server端消息:" + tm.getText());
    	} catch (Exception e) {
    		e.printStackTrace();
    	}
    }
    


    其余部分代码(没啥花头,就是sender里带了一个messageListener):

    producer.send(message);
    


    点对点,应答式消息核心代码-生产端所有代码


    package webpoc.mq.dual;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.commons.lang.RandomStringUtils;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    
    public class Client implements MessageListener {
    
    
    	public void onMessage(Message message) {
    		TextMessage tm = (TextMessage) message;
    		try {
    			System.out.println("Client接收Server端消息:" + tm.getText());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    public void start(ApplicationContext context) {
    		ConnectionFactory factory = null;
    		Connection conn = null;
    		Destination destination = null;
    		Destination replyDest = null;
    		Session session = null;
    		MessageProducer producer = null;
    		MessageConsumer consumer = null;
    		try {
    			destination = (Destination) context.getBean("destination");
    			replyDest = (Destination) context.getBean("replyDestination");
    			factory = (ActiveMQConnectionFactory) context.getBean("connectionFactory");
    			conn = factory.createConnection();
    			conn.start();
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			producer = session.createProducer(destination);
    			TextMessage message = session.createTextMessage("...Hello JMS!");
    			String correlationId = RandomStringUtils.randomNumeric(5);
    
    			consumer = session.createConsumer(replyDest);
    			message.setJMSReplyTo(replyDest);
    			message.setJMSCorrelationID(correlationId);
    
    			consumer.setMessageListener(this);
    
    		} catch (Exception e) {
    			String errorMessage = "JMSException while queueing HTTP JMS Message";
    			e.printStackTrace();
    		}
    }
    	public void send(ApplicationContext context) {
    		ConnectionFactory factory = null;
    		Connection conn = null;
    		Destination destination = null;
    		Destination replyDest = null;
    		Session session = null;
    		MessageProducer producer = null;
    		try {
    			destination = (Destination) context.getBean("destination");
    			replyDest = (Destination) context.getBean("replyDestination");
    			factory = (ActiveMQConnectionFactory) context.getBean("connectionFactory");
    			conn = factory.createConnection();
    			conn.start();
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			producer = session.createProducer(destination);
    			TextMessage message = session.createTextMessage("...Hello JMS!");
    			String correlationId = RandomStringUtils.randomNumeric(5);
    
    			message.setJMSReplyTo(replyDest);
    			message.setJMSCorrelationID(correlationId);
    			producer.send(message);
    			System.out.println("send 1 message");
    		} catch (Exception e) {
    			String errorMessage = "JMSException while queueing HTTP JMS Message";
    			e.printStackTrace();
    		}
    	}
    
    	public static void main(String[] args) {
    		final ApplicationContext context = new ClassPathXmlApplicationContext("classpath:/spring/activemq_dual.xml");
    		//sendWithAuto(context);
    		Client c = new Client();
    		c.start(context);
    		c.send(context);
    
    }
    


    点对点,应答式消息核心代码-消费端部分代码


    public void onMessage(Message message) {
    		System.out.println("on message");
    		try {
    			TextMessage response = this.session.createTextMessage();
    			if (message instanceof TextMessage) {
    				TextMessage txtMsg = (TextMessage) message;
    				String messageText = txtMsg.getText();
    				response.setText("服务器收到消息:" + messageText);
    				System.out.println(response.getText());
    			}
    			response.setJMSCorrelationID(message.getJMSCorrelationID());
    			producer.send(message.getJMSReplyTo(), response);
    
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    
    }
    

    1. 此处的send()方法内有两个参数,注意其用法
    2. 然后为这个消费端也加一个messageListener如:

    session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    producer = session.createProducer(replyDest);
    consumer = session.createConsumer(destination);
    consumer.setMessageListener(this);
    


    点对点,应答式消息核心代码-全部代码


    package webpoc.mq.dual;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.commons.lang.RandomStringUtils;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class Server implements MessageListener {
    	private ConnectionFactory factory = null;
    	private Connection conn = null;
    	private Destination destination = null;
    	Destination replyDest = null;
    	private Session session = null;
    	private MessageProducer producer = null;
    	private MessageConsumer consumer = null;
    
    	@Override
    	public void onMessage(Message message) {
    		System.out.println("on message");
    		try {
    			// 若有消息传送到服务时,先创建一个文本消息  
    			TextMessage response = this.session.createTextMessage();
    			// 若从客户端传送到服务端的消息为文本消息  
    			if (message instanceof TextMessage) {
    				// 先将传送到服务端的消息转化为文本消息  
    				TextMessage txtMsg = (TextMessage) message;
    				// 取得文本消息的内容  
    				String messageText = txtMsg.getText();
    				// 将客户端传送过来的文本消息进行处理后,设置到回应消息里面  
    				response.setText("服务器收到消息:" + messageText);
    				System.out.println(response.getText());
    			}
    			// 设置回应消息的关联ID,关联ID来自于客户端传送过来的关联ID  
    			response.setJMSCorrelationID(message.getJMSCorrelationID());
    			System.out.println("replyto===" + message.getJMSReplyTo());
    			// 生产者发送回应消息,目的由客户端的JMSReplyTo定义,内容即刚刚定义的回应消息  
    			producer.send(message.getJMSReplyTo(), response);
    
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    
    
    public void receive(ApplicationContext context) {
    
    		try {
    			destination = (Destination) context.getBean("destination");
    			replyDest = (Destination) context.getBean("replyDestination");
    			factory = (ActiveMQConnectionFactory) context.getBean("connectionFactory");
    			conn = factory.createConnection();
    			conn.start();
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			producer = session.createProducer(replyDest);
    			consumer = session.createConsumer(destination);
    			consumer.setMessageListener(this);
    		} catch (Exception e) {
    			String errorMessage = "JMSException while queueing HTTP JMS Message";
    			e.printStackTrace();
    		}
    	}
    
    	public static void main(String[] args) {
    		final ApplicationContext context = new ClassPathXmlApplicationContext("classpath:/spring/activemq_dual.xml");
    		Server s = new Server();
    		s.receive(context);
    
    	}
    
    }
    


    点对点,应答式消息核心代码-演示






  • 相关阅读:
    周记(第六周)
    周记(第五周)
    周记(第四周)
    周记(第三周)
    周记(第二周)
    《大道至简》读后感
    __proto__
    Object.prototype
    Object.setPrototypeOf(obj, proto)
    Object.getPrototypeOf(obj)
  • 原文地址:https://www.cnblogs.com/aiwz/p/6154590.html
Copyright © 2011-2022 走看看