Activemq 结合Spring应用2

之前虽然谢过类似的代码，但是脑子里还有那么一点不清楚，主要消费者生产者的感念，今天工作不忙，再次复习，那个点还是没过去，请教同事一点通了，呵呵！以前的理解还是有误差的，不过发表的文章没问题！
消费者：

activemq-consumer.xml配置如下：
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

	xsi:schemaLocation="http://www.springframework.org/schema/beans    
           http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
           
           http://activemq.apache.org/schema/core            		   
 		   http://activemq.apache.org/schema/core/activemq-core.xsd">

	<bean id="queueJmsConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616?wireFormat.maxInactivityDuration=0" />
		<property name="useAsyncSend" value="true" />
	</bean>	
	
	<!--  ActiveMQ destinations  -->

	<!--  使用Queue方式-->
	<amq:queue name="QUEUE0" physicalName="JMS-TEST-QUEUE0" />
	<amq:queue name="QUEUE1" physicalName="JMS-TEST-QUEUE1" />
	<amq:queue name="QUEUE2" physicalName="JMS-TEST-QUEUE2" />

	<!-- converter  -->
	<bean id="defaultMessageConverter" class="com.activemq.common.DefaultMessageConverter" />

	<!--  Message Driven POJO (MDP) -->
	<!-- consumer for queue -->
	<bean id="queueConsumer0" class="com.activemq.consumer.QueueConsumer" scope="prototype"/>
	<bean id="queueConsumer1" class="com.activemq.consumer.QueueConsumer" scope="prototype"/>
	<bean id="queueConsumer2" class="com.activemq.consumer.QueueConsumer2" scope="prototype"/>

	<!-- Message Listener for  -->
	<bean id="queueListener0"
		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg ref="queueConsumer0" />
		<!--  may be other method -->
		<property name="defaultListenerMethod" value="receive" />
		<!-- custom MessageConverter define -->
		<property name="messageConverter" ref="defaultMessageConverter" />
	</bean>
	<bean id="queueListener1"
		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg ref="queueConsumer1" />
		<!--  may be other method -->
		<property name="defaultListenerMethod" value="receive" />
		<!-- custom MessageConverter define -->
		<property name="messageConverter" ref="defaultMessageConverter" />
	</bean>
	<bean id="queueListener2"
		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<constructor-arg ref="queueConsumer2" />
		<!--  may be other method -->
		<property name="defaultListenerMethod" value="receive" />
		<!-- custom MessageConverter define -->
		<property name="messageConverter" ref="defaultMessageConverter" />
	</bean>
	
	<!--  listener container，MDP无需实现接口 -->
	<bean id="queueListenerContainer0"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="queueJmsConnectionFactory" />
		<property name="destination" ref="QUEUE0" />
		<property name="messageListener" ref="queueListener0" />
		<property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE" />
		<property name="receiveTimeout" value="20000" />
	</bean>
	<bean id="queueListenerContainer1"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="queueJmsConnectionFactory" />
		<property name="destination" ref="QUEUE1" />
		<property name="messageListener" ref="queueListener1" />
		<property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE" />
		<property name="receiveTimeout" value="20000" />
	</bean>
	<bean id="queueListenerContainer2"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="queueJmsConnectionFactory" />
		<property name="destination" ref="QUEUE2" />
		<property name="messageListener" ref="queueListener2" />
		<property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE" />
		<property name="receiveTimeout" value="20000" />
	</bean>
</beans>
相应的QueueConsumer类代码（自定义的）
package com.activemq.consumer;



import org.apache.log4j.Logger;

import com.activemq.common.FooMessage;

/**
 * Date: 2008-8-28
 * Time: 17:10:34
 */
public class QueueConsumer {

	int i;
	
	Logger logger=Logger.getLogger(QueueConsumer.class);
	
    public void receive(FooMessage message) {
    	i++;    	
		System.out.println("*************************************** Queue : " + message.getId()+","+System.currentTimeMillis());
    	logger.debug("Queue: " + message.getId()+"-------count"+i+"------------------------------------------------>"+System.currentTimeMillis());
    	/*try {
			Thread.sleep(60000);
		} catch (InterruptedException e) {
			logger.error("time out..............................................");
			e.printStackTrace();
		}*/
	}
    
}

ok接着就可以测试了，测试类：
package com.activemq.consumer;

import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class ConsumerTestMain {
	public static void main(String[] args) {
		BeanFactory bf = new ClassPathXmlApplicationContext("classpath:activemq-consumer.xml");
	}
}

查看消息的发送情况
启动activemq服务器，打开http://localhost:8161/admin/选择QUEUE查看
你会发现我们启动了1个消费者，再次运行测试类消费者变为2，然后群殴们启动生成者



生产者的相关配置
activemq-producer.xml

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

	xsi:schemaLocation="http://www.springframework.org/schema/beans    
           http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
           
           http://activemq.apache.org/schema/core            		   
 		   http://activemq.apache.org/schema/core/activemq-core.xsd">	
		
	<bean id="queueJmsConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616?wireFormat.maxInactivityDuration=0"/>
		<property name="useAsyncSend" value="true"/>	
	</bean>	

	<!--  ActiveMQ destinations  -->

	<!--  使用Queue方式-->
	<amq:queue name="QUEUE0" physicalName="JMS-TEST-QUEUE0" />
	<amq:queue name="QUEUE1" physicalName="JMS-TEST-QUEUE1" />
	<amq:queue name="QUEUE2" physicalName="JMS-TEST-QUEUE2" />	

	<!--  Spring JmsTemplate config -->
	<bean id="queueJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory">
			<!--  lets wrap in a pool to avoid creating a connection per send -->
			<bean class="org.springframework.jms.connection.SingleConnectionFactory">
				<property name="targetConnectionFactory" ref="queueJmsConnectionFactory" />
			</bean>
		</property>
		<!-- custom MessageConverter -->
		<property name="messageConverter" ref="defaultMessageConverter" />
		<property name="receiveTimeout" value="20000"/>
	</bean>		
	
	<!-- converter  -->
	<bean id="defaultMessageConverter" class="com.activemq.common.DefaultMessageConverter" />	

	<bean id="queueMessageProducer0" class="com.activemq.producer.QueueMessageProducer">
		<property name="template" ref="queueJmsTemplate" />
		<property name="destination" ref="QUEUE0" />
	</bean>
	<bean id="queueMessageProducer1" class="com.activemq.producer.QueueMessageProducer">
		<property name="template" ref="queueJmsTemplate" />
		<property name="destination" ref="QUEUE1" />
	</bean>
	<bean id="queueMessageProducer2" class="com.activemq.producer.QueueMessageProducer">
		<property name="template" ref="queueJmsTemplate" />
		<property name="destination" ref="QUEUE2" />
	</bean>	
</beans>

生成者类：QueueMessageProducer
package com.activemq.producer;


import javax.jms.Queue;

import org.springframework.jms.core.JmsTemplate;

import com.activemq.common.FooMessage;

public class QueueMessageProducer extends Thread {

	private Queue destination;

	private String qmid;

	private JmsTemplate template;

	public Queue getDestination() {
		return destination;
	}

	public String getQmid() {
		return qmid;
	}

	public JmsTemplate getTemplate() {
		return template;
	}

	@Override
	public void run() {
		for (int i = 0; i < 1000000; i++) {
			FooMessage fMessage = new FooMessage();
			fMessage.setId(this.getQmid() + "-" + i);
			this.send(fMessage);
		}
	}

	public void send(FooMessage message) {
		this.getTemplate().convertAndSend(this.getDestination(), message);
	}

	public void setDestination(Queue destination) {
		this.destination = destination;
	}

	public void setQmid(String qmid) {
		this.qmid = qmid;
	}

	public void setTemplate(JmsTemplate template) {
		this.template = template;
	}
}
测试类：
package com.activemq.producer;


import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class ProducerTestMain {
	public static void main(String[] args) {
		BeanFactory bf = new ClassPathXmlApplicationContext("classpath:activemq-producer.xml");
		QueueMessageProducer[] qms={
				(QueueMessageProducer)bf.getBean("queueMessageProducer0"),
				(QueueMessageProducer)bf.getBean("queueMessageProducer1"),
				(QueueMessageProducer)bf.getBean("queueMessageProducer2")};
		for (int i = 0; i < qms.length; i++) {			
			qms[i].setQmid(i+"");
			qms[i].start();
		}
	}
}

启动测试类把你的生产者类中的run方法的循环次数的数字写大些比如向我这里一样写成10万，对应的你启动的activemq服务器你就明白了
不明白的可以喝我沟通，我们共同学习，以防我们学习过程中的不足
另外我要向大家说下我的误解：
<!--  使用Queue方式-->  
19.    <amq:queue name="QUEUE0" physicalName="JMS-TEST-QUEUE0" />  
20.    <amq:queue name="QUEUE1" physicalName="JMS-TEST-QUEUE1" />  
21.    <amq:queue name="QUEUE2" physicalName="JMS-TEST-QUEUE2" />       一直以来我误以为这段代码我们创建了三个消费者其实是消费者的三个消费管道，消费者要通过这样的管道接受消费者发送的消息！