消息中间件activemq的helloworld程序 ( topic )

pom依赖

     <!--  activemq  所需要的jar 包-->
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-all</artifactId>
      <version>5.15.9</version>
    </dependency>

    <!--  activemq 和 spring 整合的基础包 -->
    <dependency>
      <groupId>org.apache.xbean</groupId>
      <artifactId>xbean-spring</artifactId>
      <version>3.16</version>
    </dependency>

生产者

import javax.jms.Connection;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicP {
	public static final String ACTIVEMQ_URL = "tcp://192.168.166.138:61616";
	public static final String TOPIC_NAME = "topic";

	public static void main(String[] args) throws Exception {

		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
		Connection connection = activeMQConnectionFactory.createConnection();
		connection.start();
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		Topic topic = session.createTopic(TOPIC_NAME);
		MessageProducer messageProducer = session.createProducer(topic);
		for (int i = 1; i < 4; i++) {
			TextMessage textMessage = session.createTextMessage("msg--" + i);
			messageProducer.send(textMessage);
		}
		messageProducer.close();
		session.close();
		connection.close();
		// session.commit();
		System.out.println("  **** 消息发送到MQ完成 ****");
	}
}

消费者

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;

public class TopicC  {
	public static final String ACTIVEMQ_URL = "tcp://192.168.166.138:61616";
    public static final String TOPIC_NAME = "topic";

	public static void main(String[] args) throws Exception {
		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
		RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
		redeliveryPolicy.setMaximumRedeliveries(3);
		// 当你第四次消费消息的时候就会将消息加入死信队列
		activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);

		Connection connection = activeMQConnectionFactory.createConnection();

		connection.start();

		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		Topic topic = session.createTopic(TOPIC_NAME);
		MessageConsumer messageConsumer = session.createConsumer(topic);
		/**
		 * 第一种方法
		while(true) {
			TextMessage t=(TextMessage) messageConsumer.receive();	
			if(t!=null) {
				System.out.println(t.getText());
			}else{
				break;
			}
		}
		**/
		messageConsumer.setMessageListener(new MessageListener() {
			public void onMessage(Message message) {
				try {
					if (null != message && message instanceof TextMessage) {
						TextMessage textMessage = (TextMessage) message;
						System.out.println("****cluster msg：" + textMessage.getText());
					}
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}); 
		System.in.read();
		messageConsumer.close();
		session.close();
		connection.close();
		System.out.println("///");
	}
}