activemq queue模式与 subscribe/publ

作者: John_Phil | 来源:发表于2019-04-12 22:13 被阅读0次

activemq queue模式与 subscribe/publ
消息队列ActiveMQ的初次使用
activemq消息流转解读
springboot activemq（一）Queue(p2p)
ActiveMQ的消息模式——队列模式（Queue）
用数组模拟队列（Java）
RocketMQ
复习下MQ
ActiveMQ&RocketMQ
ActiveMQ持久化方案

P2P （点对点）消息域使用 queue 作为 Destination，消息可以被同步或异步的发送和接收，每个消息只会给一个 Consumer 传送一次。
Consumer 可以使用 MessageConsumer.receive() 同步地接收消息，也可以通过使用MessageConsumer.setMessageListener() 注册一个 MessageListener 实现异步接收。
多个 Consumer 可以注册到同一个 queue 上，但一个消息只能被一个 Consumer 所接收，然后由该 Consumer 来确认消息。并且在这种情况下，Provider 对所有注册的 Consumer 以轮询的方式发送消息。

queue

在maven项目中引入jar包

maven-repository

    <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all -->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.14.1</version>
        </dependency>

activemq参考api http://activemq.apache.org/maven/

image.png

创建producer 数据提供方

image.png

package com.neusoft.providers;

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

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

public class P2PProducer1 {
    //默认连接用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    //默认连接密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    //默认连接地址
    private static final String BROKEURL = "tcp://192.168.110.134:61616";
    //发送的消息数量
    private static final int SENDNUM = 30;
    public static void main(String[] args) {
        //连接工厂
        ConnectionFactory connectionFactory;
        //连接
        Connection connection = null;
        //会话 接受或者发送消息的线程
        Session session;
        //消息的目的地
        Destination destination;
        //消息生产者
        MessageProducer messageProducer=null;
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(P2PProducer1.USERNAME,     
         P2PProducer1.PASSWORD, P2PProducer1.BROKEURL);
        try {
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //创建一个名称为consumer a baozi的消息队列
            destination = session.createQueue("consumer a baozi");
            //创建消息生产者
            messageProducer = session.createProducer(destination);
            //发送消息
            sendMessage(session, messageProducer);
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }

    }
    /**
     * 发送消息
     * @param session
     * @param messageProducer  消息生产者
     * @throws Exception
     */
    public static void sendMessage(Session session,MessageProducer messageProducer) throws Exception{
        for (int i = 0; i < P2PProducer1.SENDNUM; i++) {
            //创建一条文本消息
            TextMessage message = session.createTextMessage("baozi" +i);
            System.out.println("发送消息：baozi" + i);
            //通过消息生产者发出消息
            messageProducer.send(message);
        }
    }
}

运行共计发送消息30条

image.png

查看activemq，有三十个包子（三十条消息）待消费。

image.png

查看topic可以看到注册及连接情况

查看topic
信息对照表

image.png

consumer消费者：

image.png

package com.neusoft.consumers;


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

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

public class P2PConsumer1 {
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
    private static final String BROKEURL = "tcp://192.168.110.134:61616";//默认连接地址

    public static void main(String[] args) {
        ConnectionFactory connectionFactory;//连接工厂
        Connection connection ;//连接
        Session session;//会话 接受或者发送消息的线程
        Destination destination;//消息的目的地
        MessageConsumer messageConsumer;//消息的消费者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(P2PConsumer1.USERNAME, P2PConsumer1.PASSWORD, P2PConsumer1.BROKEURL);
        try {
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //创建一个连接consumer a baozi的消息队列
            destination = session.createQueue("consumer a baozi");
            //创建消息消费者
            messageConsumer = session.createConsumer(destination);
            while (true) {
                TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);//接受超时间隔（100000ms）内的消息    100000ms为过期时间值
                if(textMessage != null){
                    System.out.println("收到的消息:" + textMessage.getText());
                }else {
                    break;
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

image.png

此时会发现 mq的消息队列依次被consumer使用 activemq中待消费数据为0，而消息处理过得总记录在 Messages Enqueued | Messages Dequeued 列下

image.png

Pub/Sub（发布/订阅，Publish/Subscribe）消息域使用 topic 作为 Destination，发布者向 topic 发送消息，订阅者注册接收来自 topic 的消息。发送到 topic 的任何消息都将自动传递给所有订阅者。接收方式（同步和异步）与 P2P 域相同。
除非显式指定，否则 topic 不会为订阅者保留消息。当然，这可以通过持久化（Durable）订阅来实现消息的保存。这种情况下，当订阅者与 Provider 断开时，Provider 会为它存储消息。当持久化订阅者重新连接时，将会收到所有的断连期间未消费的消息。

Pub/Sub

消费者：

package com.neusoft.consumers;
import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;
public class TopicConsumer2 {

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

    public static void TestTopicConsumer() throws Exception{
        //1、创建工厂连接对象，需要制定ip和端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.110.134:61616");
        //2、使用连接工厂创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        //3、开启连接
        connection.start();
        //4、使用连接对象创建会话（session）对象  
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
        Topic topic = session.createTopic("test-topic");
        //6、使用会话对象创建生产者对象
        MessageConsumer consumer = session.createConsumer(topic);
        //7、向consumer对象中设置一个messageListener对象，用来接收消息
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                // TODO Auto-generated method stub
                if(message instanceof TextMessage){
                    TextMessage textMessage = (TextMessage)message;
                    try {
                        System.out.println(textMessage.getText());
                    } catch (JMSException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        });
        //8、程序等待接收用户消息
        System.in.read();
        //9、关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}

生产者：

package com.neusoft.providers;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class TopicProducer2 {

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

    public static void TestTopicProducer() throws Exception {
        //1、创建工厂连接对象，需要制定ip和端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.110.134:61616");
        //2、使用连接工厂创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        //3、开启连接
        connection.start();
        //4、使用连接对象创建会话（session）对象  ，第一个参数是否支持事务
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
        Topic topic = session.createTopic("test-topic");
        //6、使用会话对象创建生产者对象
        MessageProducer producer = session.createProducer(topic);
        //7、使用会话对象创建一个消息对象
        TextMessage textMessage = session.createTextMessage("hello!test-topic");
        //8、发送消息
        producer.send(textMessage);
        //9、关闭资源
        producer.close();
        session.close();
        connection.close();
    }
}

先订阅再发布
先运行consumer 再运行 producer

运行consumer

producer

信息对照表

image.png
consumer接收到 mq发送的信息

image.png

activemq queue模式与 subscribe/publ
P2P （点对点）消息域使用 queue 作为 Destination，消息可以被同步或异步的发送和接收，每个消息...
消息队列ActiveMQ的初次使用
＃消息队列ActiveMQ的初次使用一个简单的整合案例, 初步实现 activemq 的 queue 队列模式和...
activemq消息流转解读
activemq教程极客学院 Queue与Topic的比较1.JMS Queue执行load balancer语...
springboot activemq（一）Queue(p2p)
springboot activemq（一）Queue(p2p)和Topic(发布订阅) 这里安装activemq...
ActiveMQ的消息模式——队列模式（Queue）
一、队列模式特点客户端包括生产者和消费者队列中的消息只能被一个消费者消费消费者可以随时消费队列中的消息二、...
用数组模拟队列（Java）
package com.atguigu.queue; import java.util.Scanner; publ...
RocketMQ
Message Queue 常用：RocketMQ、RabbitMQ（金融）、ActiveMQ（Apache开源）...
复习下MQ
常见的几种mq: 1.activemq：java自己写的，工作模式单一只有queue（多对一）和topics（广播...
ActiveMQ&RocketMQ
ActiveMQ activemq admin 中的subject是指主题（1发N收），queue是队列（1发1收...
ActiveMQ持久化方案
1 ActiveMQ持久化模型 1.1 PTP Queue的存储是很简单的，就是一个FIFO的Queue 1.2 ...