黑马程序员技术交流社区

标题: 三、 activemq 入门级代码 [打印本页]

---

作者: y6814365    时间: 2018-5-11 10:57
标题: 三、 activemq 入门级代码
1.queue (session会话 )
请看第二篇文章https://blog.csdn.net/Mr_yangzc/article/details/80060754
1. 生产者

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
* @author Yang
* @create 2018-04-19 17:11
* 描述： 生产者
*/
public class Producer {

    /**
     * 用户名
     */
    private static final String userName = ActiveMQConnection.DEFAULT_USER;
    /**
     * 密码
     */
    private static final String passWord = ActiveMQConnection.DEFAULT_PASSWORD;
    /**
     * url
     */
    private static final String brokerUrl = ActiveMQConnection.DEFAULT_BROKER_URL;

    public void send(String message) {


        try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, passWord, brokerUrl);

            final Connection connection = connectionFactory.createConnection();

            connection.start();
          // true 为开启会话 需要提交
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            //创建队列
            final Queue queue = session.createQueue("test");
            final MessageProducer producer = session.createProducer(queue);

            TextMessage textMessage = session.createTextMessage(message);

            //发送
            producer.send(textMessage);
// 开启会话 需要提交
// session.commit(); producer.close(); session.close(); connection.close(); } catch (JMSException e) { e.printStackTrace(); } } public static void main(String[] args) { Producer producer = new Producer(); producer.send("hello world"); }}


2.消费者

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
* @author Yang
* @create 2018-04-19 17:10
* 描述： 消费者
*/

public class Consumer {

    /**
     * 用户名
     */
    private static final String userName = ActiveMQConnection.DEFAULT_USER;
    /**
     * 密码
     */
    private static final String passWord = ActiveMQConnection.DEFAULT_PASSWORD;
    /**
     * url
     */
    private static final String brokerUrl = ActiveMQConnection.DEFAULT_BROKER_URL;

    public void receive() {


        try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, passWord, brokerUrl);

            final Connection connection = connectionFactory.createConnection();

            connection.start();

            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            //创建队列
            final Queue queue = session.createQueue("test");

            final MessageConsumer messageConsumer = session.createConsumer(queue);


            messageConsumer.setMessageListener(n -> {

                try {
                    TextMessage msg = (TextMessage) n;

                    final String text = msg.getText();

                    if (text.equalsIgnoreCase("hello world")) {
                        System.out.println("               接受信息：         " + msg.getText());
                    } else {//默认为6次
                        System.out.println("     测试重发次数 " );
                        int i = 1 / 0;
                    }


                } catch (JMSException e) {
//                    e.printStackTrace();
                }

            });

        } catch (JMSException e) {
//            e.printStackTrace();
        }

    }

    public static void main(String[] args) {
        Consumer consumer = new Consumer();
        consumer.receive();
    }


2.topic  首先要启动 消费者进行订阅
1.producer 生产者

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
* @author Yang
* @create 2018-04-19 17:11
* 描述： 生产者
*/
public class TopicProducer {

    /**
     * 用户名
     */
    private static final String userName = ActiveMQConnection.DEFAULT_USER;
    /**
     * 密码
     */
    private static final String passWord = ActiveMQConnection.DEFAULT_PASSWORD;
    /**
     * url
     */
    private static final String brokerUrl = ActiveMQConnection.DEFAULT_BROKER_URL;

    public void send(String message) {


        try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, passWord, brokerUrl);

            final Connection connection = connectionFactory.createConnection();

            connection.start();

            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);


            //创建队列
            final Topic topic = session.createTopic("topic-test");
            final MessageProducer producer = session.createProducer(topic);

            TextMessage textMessage = session.createTextMessage(message);

            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
//            producer.setTimeToLive(10);

//            发送
            producer.send(textMessage);
//            producer.send(textMessage, DeliveryMode.PERSISTENT, 1, 60 * 60 * 24);

//            session.commit();

            producer.close();

            session.close();

            connection.close();

        } catch (JMSException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) {
        TopicProducer producer = new TopicProducer();
        producer.send("hello world");

    }


2.consumer 消费者

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
* @author Yang
* @create 2018-04-19 17:10
* 描述： 消费者
*/


public class TopicConsumer {

    /**
     * 用户名
     */
    private static final String userName = ActiveMQConnection.DEFAULT_USER;
    /**
     * 密码
     */
    private static final String passWord = ActiveMQConnection.DEFAULT_PASSWORD;
    /**
     * url
     */
    private static final String brokerUrl = ActiveMQConnection.DEFAULT_BROKER_URL;

    public void receive() {


        try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, passWord, brokerUrl);

            final Connection connection = connectionFactory.createConnection();
             //设置客户端id
          //  connection.setClientID("client-1");

            connection.start();

            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            //创建队列
            final Topic topic = session.createTopic("topic-test");

            final MessageConsumer messageConsumer = session.createConsumer(topic);//普通订阅
//            MessageConsumer consumer = session.createDurableSubscriber(topic,"bb"); //持久订阅

            messageConsumer.setMessageListener(n -> {

                try {
                    TextMessage msg = (TextMessage) n;

                    final String text = msg.getText();

                    if (text.equalsIgnoreCase("hello world")) {
                        System.out.println("               接受信息：         " + msg.getText());
                    } else {
                        System.out.println("     测试重发次数 ");
                        int i = 1 / 0;
                    }


                } catch (JMSException e) {
//                    e.printStackTrace();
                }

            });

        } catch (JMSException e) {
//            e.printStackTrace();
        }

    }

    public static void main(String[] args) {
        TopicConsumer consumer = new TopicConsumer();
        consumer.receive();
    }


}

---

欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2