JMS-activeMq发布订阅模式(非持久订阅)

Publisher的代码:

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Publisher {

    // 单例模式

    // 1、连接工厂
    private ConnectionFactory connectionFactory;
    // 2、连接对象
    private Connection connection;
    // 3、Session对象
    private Session session;
    // 4、生产者
    private MessageProducer messageProducer;

    public Publisher() {

        try {
            this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
                    "123", "tcp://localhost:61616");
            this.connection = connectionFactory.createConnection();
            this.connection.start();
            // 不使用事务
            // 设置客户端签收模式
            this.session = this.connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            this.messageProducer = this.session.createProducer(null);
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }

    }

    public Session getSession() {
        return this.session;
    }

    public void send1(/* String QueueName, Message message */) {
        try {

            Destination destination = this.session.createTopic("topic1");
            MapMessage msg1 = this.session.createMapMessage();
            msg1.setString("name", "张三");
            msg1.setInt("age", 22);

            MapMessage msg2 = this.session.createMapMessage();
            msg2.setString("name", "李四");
            msg2.setInt("age", 25);

            MapMessage msg3 = this.session.createMapMessage();
            msg3.setString("name", "张三");
            msg3.setInt("age", 30);

            // 发送消息到topic1
            this.messageProducer.send(destination, msg1,
                    DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
            this.messageProducer.send(destination, msg2,
                    DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
            this.messageProducer.send(destination, msg3,
                    DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);

        } catch (JMSException e) {
            throw new RuntimeException(e);
        }
    }

    public void send2() {
        try {
            Destination destination = this.session.createTopic("topic1");
            TextMessage message = this.session.createTextMessage("我是一个字符串");
            // 发送消息
            this.messageProducer.send(destination, message,
                    DeliveryMode.NON_PERSISTENT, 4, 1000 * 60 * 10);
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }

    }

    public static void main(String[] args) {
        Publisher producer = new Publisher();
        producer.send1();

    }

}

Subscribe的代码:

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Subscriber {

    // 单例模式

    // 1、连接工厂
    private ConnectionFactory connectionFactory;
    // 2、连接对象
    private Connection connection;
    // 3、Session对象
    private Session session;
    // 4、生产者
    private MessageConsumer messageConsumer;
    // 5、目的地址
    private Destination destination;

    public Subscriber() {

        try {
            this.connectionFactory = new ActiveMQConnectionFactory("zhangsan",
                    "123", "tcp://localhost:61616");
            this.connection = connectionFactory.createConnection();
            this.connection.start();
            // 不使用事务
            // 设置客户端签收模式
            this.session = this.connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            this.destination = this.session.createTopic("topic1");
            this.messageConsumer = this.session.createConsumer(destination);
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }

    }

    public Session getSession() {
        return this.session;
    }

    // 用于监听消息队列的消息
    class MyLister implements MessageListener {

        @Override
        public void onMessage(Message message) {
            try {
                if (message instanceof TextMessage) {

                }
                if (message instanceof MapMessage) {
                    MapMessage ret = (MapMessage) message;
                    System.out.println(ret.toString());
                    System.out.println(ret.getString("name"));
                    System.out.println(ret.getInt("age"));
                    // 因为设置的是客户端的签收模式，所以要手动的去确认消息的消费
                    message.acknowledge();
                }
            } catch (JMSException e) {
                throw new RuntimeException(e);
            }
        }

    }

    // 用于异步监听消息
    public void receiver() {
        try {
            this.messageConsumer.setMessageListener(new MyLister());
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) {
        Subscriber conmuser = new Subscriber();
        conmuser.receiver();

    }

}

 

 先启动消费者(先订阅后消费)，再启动发布者