activemq两种实现方式

2018-06-18 02:53:37来源：未知阅读 ()

第一种：点对点

#发布者
public class Producer {
    
    private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
    private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
    private static final String brokerURL = "tcp://192.168.178.X:61616";

    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂类
        ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.启动连接
        connection.start();
        
        //4.创建会话对象session(事务transacted为true,参数2不生效)
        //acknowledgeMode:
        Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
        
        //5.目的地
        Queue queue = session.createQueue("mq-test-01");
        //7.创建发送者
        MessageProducer producer = session.createProducer(queue);
        for(int i=1;i<=10;i++) {
            //6.消息对象
            TextMessage message = session.createTextMessage();
            message.setText("消息"+i);
            //8.发送消息
            producer.send(message);
        }
        //9.会话提交
//        session.commit();
        
        //10.关闭连接
        connection.close();
    }
}

#消费者
public class Consumer1 {
    
    private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
    private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
    private static final String brokerURL = "tcp://192.168.178.X:61616";

    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂类
        ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.启动连接
        connection.start();
        
        //4.创建会话对象session(事务transacted为true,参数2不生效)
        //acknowledgeMode:
        Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
        
        //5.目的地
        Queue queue = session.createQueue("mq-test-01");
        
        //6.接收消息对象
        MessageConsumer consumer = session.createConsumer(queue);
        
        //7.通过监听器接收消息
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage)message;
                //获取消息
                try {
                    String msg = textMessage.getText();
                    System.out.println(msg);
                } catch (JMSException e) {
                }
            }
        });
        
    }
}

第二种：发布者/订阅者

启动顺序：先订阅、再发布

#订阅者
public class Subscriber1 {
    
    private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
    private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
    private static final String brokerURL = "tcp://192.168.129.10:61616";

    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂类
        ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.启动连接
        connection.start();
        
        //4.创建会话对象session(事务transacted为true,参数2不生效)
        //acknowledgeMode:
        Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
        
        //5.目的地
        Topic topic = session.createTopic("mq-test-02");
        
        //6.接收消息对象
        MessageConsumer consumer = session.createConsumer(topic);
        
        //7.通过监听器接收消息
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage)message;
                //获取消息
                try {
                    String msg = textMessage.getText();
                    System.out.println(msg);
                } catch (JMSException e) {
                }
            }
        });
        
    }
}

#发布者
public class Publisher {
    
    private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER;
    private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD;
    private static final String brokerURL = "tcp://192.168.129.10:61616";

    public static void main(String[] args) throws JMSException {
        //1.创建连接工厂类
        ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL);
        //2.创建连接
        Connection connection = factory.createConnection();
        //3.启动连接
        connection.start();
        
        //4.创建会话对象session(事务transacted为true,参数2不生效)
        //acknowledgeMode:
        Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
        
        //5.目的地
        Topic topic = session.createTopic("mq-test-02");
        //7.创建发送者
        MessageProducer producer = session.createProducer(topic);
        for(int i=1;i<=10;i++) {
            //6.消息对象
            TextMessage message = session.createTextMessage();
            message.setText("消息"+i);
            //8.发送消息
            producer.send(message);
            //设置自动持久化
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        }
        //9.会话提交
//        session.commit();
        
        //10.关闭连接
        connection.close();
    }
}