接上篇 JMS笔记(一)，启动ActiveMQ后，打开http://127.0.0.1:8161/admin管理界面，用户名admin密码admin，点击上面的Queues菜单，创建一个q_test_20160108，之后就可以向这个queue写东西了

1. P2P模式

生产者示例代码：

　　@Test

    public void testActiveMqSend(){

        // ConnectionFactory ：连接工厂，JMS 用它创建连接

        ConnectionFactory connectionFactory;

        // Connection ：JMS 客户端到JMS Provider 的连接

        Connection connection = null;

        // Session： 一个发送或接收消息的线程

        Session session;

        // Destination ：消息的目的地;消息发送给谁.

        Destination destination;

        // MessageProducer：消息发送者

        MessageProducer producer;

        // TextMessage message;

        // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar

        connectionFactory = new ActiveMQConnectionFactory(

                ActiveMQConnection.DEFAULT_USER,

                ActiveMQConnection.DEFAULT_PASSWORD,

                "tcp://127.0.0.1:61616");

        try {

            // 构造从工厂得到连接对象

            connection = connectionFactory.createConnection();

            // 启动

            connection.start();

            // 获取操作连接

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

            // 获取session注意参数值q_test_20160108是一个服务器的queue，须在在ActiveMq的console配置

            destination = session.createQueue("q_test_20160108");

            // 得到消息生成者【发送者】

            producer = session.createProducer(destination);

            // 设置不持久化，实际根据项目决定

            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // 构造消息，

            sendMessage(session, producer);

            session.commit();

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            try {

                if (null != connection)

                    connection.close();

            } catch (Throwable ignore) {

            }

        }

    }

private static final int SEND_NUMBER = 5;

public static void sendMessage(Session session, MessageProducer producer)

        throws Exception {

    for (int i = 1; i <= SEND_NUMBER; i++) {

        TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i);

        // 发送消息到目的地方

        System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);

        producer.send(message);

    }

}

运行完以上代码去管理界面查看q_test_20160108

消费者示例代码：

　　@Test

    public void testActiveMqReceive(){

        // ConnectionFactory ：连接工厂，JMS 用它创建连接

        ConnectionFactory connectionFactory;

        // Connection ：JMS 客户端到JMS Provider 的连接

        Connection connection = null;

        // Session： 一个发送或接收消息的线程

        Session session;

        // Destination ：消息的目的地;消息发送给谁.

        Destination destination;

        // 消费者，消息接收者

        MessageConsumer consumer;

        connectionFactory = new ActiveMQConnectionFactory(

                ActiveMQConnection.DEFAULT_USER,

                ActiveMQConnection.DEFAULT_PASSWORD,

                "tcp://127.0.0.1:61616");

        try {

            // 构造从工厂得到连接对象

            connection = connectionFactory.createConnection();

            // 启动

            connection.start();

            // 获取操作连接

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

            // 获取session注意参数值q_test_20160108是一个服务器的queue，须在在ActiveMq的console配置

            destination = session.createQueue("q_test_20160108");

            consumer = session.createConsumer(destination);

            while (true) {

                //设置接收者接收消息的时间，为了便于测试，这里设定为100s

                TextMessage message = (TextMessage) consumer.receive(100000);

                if (null != message) {

                    System.out.println("收到消息" + message.getText());

                } else {

                    break;

                }

            }

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            try {

                if (null != connection)

                    connection.close();

            } catch (Throwable ignore) {

            }

        }

    }

以上代码去mq取消息，运行完之后，再去管理界面查看q_test_20160108，刚刚的消息没有了,但是能够看到统计，有多少消息进来，多少消息出去，当前还有多少消息

，当前多少消费者等

2.  pub/sub模式

发布者示例：

 　　@Test

    public void testActiveMqSend(){

        // ConnectionFactory ：连接工厂，JMS 用它创建连接

        ConnectionFactory connectionFactory;

        // Connection ：JMS 客户端到JMS Provider 的连接

        Connection connection = null;

        // Session： 一个发送或接收消息的线程

        Session session;

        // Destination ：消息的目的地;消息发送给谁.

        Destination destination;

        // MessageProducer：消息发送者

        MessageProducer producer;

        // TextMessage message;

        // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar

        connectionFactory = new ActiveMQConnectionFactory(

                ActiveMQConnection.DEFAULT_USER,

                ActiveMQConnection.DEFAULT_PASSWORD,

                "tcp://127.0.0.1:61616");

        try {

            // 构造从工厂得到连接对象

            connection = connectionFactory.createConnection();

            // 启动

            connection.start();

            // 获取操作连接

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

            destination = session.createTopic("topic_1");

            // 得到消息生成者【发送者】

            producer = session.createProducer(destination);

            // 设置不持久化，此处学习，实际根据项目决定

            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // 构造消息，此处写死，项目就是参数，或者方法获取

            sendMessage(session, producer);

            session.commit();

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            try {

                if (null != connection)

                    connection.close();

            } catch (Throwable ignore) {

            }

        }

    }

    private static final int SEND_NUMBER = 5;

    public static void sendMessage(Session session, MessageProducer producer)

            throws Exception {

        for (int i = 1; i <= SEND_NUMBER; i++) {

            TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i);

            // 发送消息到目的地方

            System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);

            producer.send(message);

        }

    }

执行完以上语句，在管理界面点击topics菜单，可以看到topic_1

订阅者示例：

@Test

    public void testActiveMqReceive(){

        // ConnectionFactory ：连接工厂，JMS 用它创建连接

        ConnectionFactory connectionFactory;

        // Connection ：JMS 客户端到JMS Provider 的连接

        Connection connection = null;

        // Session： 一个发送或接收消息的线程

        Session session;

        // Destination ：消息的目的地;消息发送给谁.

        Destination destination;

        // 消费者，消息接收者

        MessageConsumer consumer;

        connectionFactory = new ActiveMQConnectionFactory(

                ActiveMQConnection.DEFAULT_USER,

                ActiveMQConnection.DEFAULT_PASSWORD,

                "tcp://127.0.0.1:61616");

        try {

            // 构造从工厂得到连接对象

            connection = connectionFactory.createConnection();
　　　　　　　//持久化订阅必须使用下语句！！！！参数是clientid

            connection.setClientID("mysub_3");

            // 启动

            connection.start();

            // 获取操作连接

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

            destination = session.createTopic("topic_1");

            //consumer = session.createConsumer(destination);普通订阅
　　　　　　 //持久化订阅！！！第二个参数是clientid

            consumer = session.createDurableSubscriber((Topic) destination,"mysub_3");

            while (true) {

                //设置接收者接收消息的时间，为了便于测试，这里设定为100s

                TextMessage message = (TextMessage) consumer.receive(100000);

                if (null != message) {

                    System.out.println("收到消息" + message.getText());

                } else {

                    break;

                }

            }

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            try {

                if (null != connection)

                    connection.close();

            } catch (Throwable ignore) {

            }

        }

    }

发现一个问题，运行以上代码，不能收到topic_1上的消息，开始一直以为有什么东西有问题，纠结了一阵子，后来想通了，对持久订阅理解不对，持久订阅从你订阅开始收到消息，离线了，然后在上线，期间的消息是可以收到了，但是你订阅之前的就不能收到了

在管理界面Subscribers菜单下可以看到订阅者列表

参考文章

1.  ActiveMQ消息队列的使用及应用

2.  消息中间件ActiveMQ

最新文章

  1. [原]使用node-mapnik生成openstreetmap-carto风格的瓦片
  2. PHP执行文档操作
  3. 使用eclipse+fiddler+微信web开发者工具调试本地微信页面
  4. java中文乱码解决方法汇总
  5. 再谈Bellman-Ford
  6. 在iis6.0公布asp.net mvc3网站
  7. Sprint第二个冲刺（第一天）
  8. Cloudera 建议使用 NTP 使 Hadoop 群集实现时间同步
  9. js页面加载进度条(这个就比较正式了，改改时间就完事儿)
  10. windows下qtcreator添加ICE库文件
  11. 运用PhantomJS测试JavaScript
  12. Kafka生产者-向Kafka中写入数据
  13. 【转】手把手教你读取Android版微信和手Q的聊天记录(仅作技术研究学习)
  14. Kafka消费者组再均衡问题
  15. http请求requestUtils
  16. tensorflow 代码阅读
  17. Unity自动切割动画
  18. LVDS时序分析
  19. 前端基础之CSS的引入+HTML标签选择器+CSS操作属性
  20. js 横屏 竖屏 相关代码 与知识点

热门文章

  1. elisp debug
  2. libpcap/wwinpcap
  3. Git Book 中文版 - Git的撤消操作 - 重置, 签出 和 撤消
  4. .NET Reactor 命令行使用
  5. 微服务架构下分布式Session管理
  6. 信息图：iOS 7开发者需要知道的事
  7. No module named BeautifulSoup
  8. 用C语言实现ipv4地址字符串是否合法
  9. Delphi实例-IdTCPServer和IdTCPClient的使用(支持文件发送)
  10. 【转】强大的vim配置文件，让编程更随意