分布式--ActiveMQ 消息中间件(一)-阿里云开发者社区

1. ActiveMQ

1). ActiveMQ

ActiveMQ是Apache所提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服务）规范。JMS是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC,它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。

2). Java Message Service(JMS)

JMS支持两种消息发送和接收模型。

一种称为P2P(Ponit to Point)模型，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输称为可能，P2P模型在点对点的情况下进行消息传递时采用。

图1.png
另一种称为Pub/Sub(Publish/Subscribe，即发布-订阅)模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为topic(主题)。主题可以认为是消息传递的中介，消息发布这将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。

图2.png

3). JMS术语

Provider/MessageProvider：生产者
Consumer/MessageConsumer：消费者
PTP：Point To Point，点对点通信消息模型
Pub/Sub：Publish/Subscribe，发布订阅消息模型
Queue：队列，目标类型之一，和PTP结合
Topic：主题，目标类型之一，和Pub/Sub结合
ConnectionFactory：连接工厂，JMS用它创建连接
Connnection：JMS Client到JMS Provider的连接
Destination：消息目的地，由Session创建
Session：会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是Session创建的

4). ActiveMQ下载

图3.png

bin (windows下面的bat(分32、64位)和unix/linux下面的sh)
conf (activeMQ配置目录，包含最基本的activeMQ配置文件）
data (默认是空的)
docs (index,replease版本里面没有文档，-.-b不知道为啥不带）
example (几个例子)
lib (activemMQ使用到的lib)
webapps 注意ActiveMQ自带Jetty提供Web管控台
webapps-demo 示例
activemq-all-5.15.3.jar
LICENSE.txt
README.txt

5). 配置

Web控制台账号和密码(apache-activemq-5.15.3\conf)

图4.png
网络端口(apache-activemq-5.15.3\conf)--默认为8161

图5.png

6). 启动

\apache-activemq-5.15.3\bin\win64\目录下双击activemq.bat文件，在浏览器中输入http://localhost:8161/admin/, 用户名和密码输入admin即可

图6.png

7). 消息中间件(MOM：Message Orient middleware)

消息中间件有很多的用途和优点：

1 将数据从一个应用程序传送到另一个应用程序，或者从软件的一个模块传送到另外一个模块；
1. 负责建立网络通信的通道，进行数据的可靠传送。
1. 保证数据不重发，不丢失
1. 能够实现跨平台操作，能够为不同操作系统上的软件集成技工数据传送服务

8).什么情况下使用ActiveMQ?

多个项目之间集成
(1) 跨平台
(2) 多语言
(3) 多项目
降低系统间模块的耦合度，解耦
(1) 软件扩展性
系统前后端隔离
(1) 前后端隔离，屏蔽高安全区

2. ActiveMQ 示例

1). P2P 示例

I. 导包--activemq-all-5.15.3.jar
II. Producer

/**
 * 定义消息的生产者
 * @author mazaiting
 */
public class Producer {
    // 用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    // 密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    // 链接
    private static final String BROKENURL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
    /**
     * 定义消息并发送，等待消息的接收者(消费者)消费此消息
     * @param args
     * @throws JMSException 
     */
    public static void main(String[] args) throws JMSException {
        // 消息中间件的链接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                USERNAME, PASSWORD, BROKENURL);
        // 连接
        Connection connection = null;
        // 会话
        Session session = null;
        // 消息的目的地
        Destination destination = null;
        // 消息生产者
        MessageProducer messageProducer = null;
        
        try {
            // 通过连接工厂获取链接
            connection = connectionFactory.createConnection();
            // 创建会话，进行消息的发送
            // 参数一：是否启用事务
            // 参数二：设置自动签收
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            // 创建消息队列
            destination = session.createQueue("talkWithMo");
            // 创建一个消息生产者
            messageProducer = session.createProducer(destination);
            // 设置持久化/非持久化， 如果非持久化，MQ重启后可能后导致消息丢失
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            // 模拟发送消息
            for (int i = 0; i < 5; i++) {
                TextMessage textMessage = session.createTextMessage("给妈妈发送的消息:"+i);
                System.out.println("textMessage: " + textMessage);
                messageProducer.send(textMessage);
            }
            
            // 如果设置了事务，会话就必须提交
            session.commit();
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            if (null != connection) {
                connection.close();
            }
        }
    }
}

III. Consumer

/**
 * 定义消息的消费者
 * @author mazaiting
 */
public class Consumer {
    // 用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    // 密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    // 链接
    private static final String BROKENURL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
    /**
     * 接收消息
     * @param args
     * @throws JMSException 
     */
    public static void main(String[] args) throws JMSException {
        // 消息中间件的链接工厂
        ConnectionFactory connectionFactory = null;
        // 链接
        Connection connection = null;
        // 会话
        Session session = null;
        // 消息的目的地
        Destination destination = null;
        // 消息的消费者
        MessageConsumer messageConsumer = null;
        // 实例化链接工厂，创建一个链接
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKENURL);
        
        try {
            // 通过工厂获取链接
            connection = connectionFactory.createConnection();
            // 启动链接
            connection.start();
            // 创建会话，进行消息的接收
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            // 创建消息队列
            destination = session.createQueue("talkWithMo");
            // 创建一个消息的消费者
            messageConsumer = session.createConsumer(destination);
            
            // 模拟接收消息
            while (true) {
                TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);
                if (null != textMessage) {
                    System.out.println("收到消息： " + textMessage);
                } else {
                    break;
                }
            }
            // 提交
            session.commit();
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            if (null != connection) {
                connection.close();
            }
        }
    }
}

IV. 测试

先运行生产者Producer

图7.png

ActiveMQ控制台

图8.png

再运行消费者Consumer

图9.png

ActiveMQ控制台

图10.png

V. 消息类型

StreamMessage Java原始值的数据流
MapMessage 一套名称-键值对
TextMessage 一个字符串对象
ObjectMessage 一个序列号的Java对象
BytesMessage 一个未解释字节的数据流
VI. 控制台 Queue
Messages Enqueued：表示生产了多少条消息，记做P
Messages Dequeued：表示消费了多少条消息，记做C
Number Of Consumers：表示在该队列上还有多少消费者在等待接受消息
Number Of Pending Messages：表示还有多少条消息没有被消费，实际上是表示消息的积压程度，就是P-C
VII. 签收
签收就是消费者接受到消息后，需要告诉消息服务器，我收到消息了。当消息服务器收到回执后，本条消息将失效。因此签收将对PTP模式产生很大影响。如果消费者收到消息后，并不签收，那么本条消息继续有效，很可能会被其他消费者消费掉！
AUTO_ACKNOWLEDGE：表示在消费者receive消息的时候自动的签收
CLIENT_ACKNOWLEDGE：表示消费者receive消息后必须手动的调用acknowledge()方法进行签收
DUPS_OK_ACKNOWLEDGE：签不签收无所谓了，只要消费者能够容忍重复的消息接受，当然这样会降低Session的开销

2). request/reply模型

I. 实现思路

图11.png

Client的Producer发出一个JMS message形式的request，request上附加了一些额外的属性：

correlation ID(用来和返回的correlation ID对比进行验证)，
JMSReplyTo属性(放置jms message的destination，这样worker的Consumer获得jms message就能得到destination)

Worker的consumer收到requset，处理request并用producer发出reply，destination就从requset的JMSReplyTo属性中得到。

II. Server代码

public class Server implements MessageListener {
    // 经纪人链接
    private static final String BROKER_URL = "tcp://localhost:61616";
    // 请求队列
    private static final String REQUEST_QUEUE = "requestQueue";
    // 经纪人服务
    private BrokerService brokerService;
    // 会话
    private Session session;
    // 生产者
    private MessageProducer producer;
    // 消费者
    private MessageConsumer consumer;
    
    private void start() throws Exception {
        createBroker();
        setUpConsumer();
    }

    /**
     * 创建经纪人
     * @throws Exception 
     */
    private void createBroker() throws Exception {
        // 创建经纪人服务
        brokerService = new BrokerService();
        // 设置是否持久化
        brokerService.setPersistent(false);
        // 设置是否使用JMX
        brokerService.setUseJmx(false);
        // 添加链接
        brokerService.addConnector(BROKER_URL);
        // 启动
        brokerService.start();
    }
    
    /**
     * 设置消费者
     * @throws JMSException 
     */
    private void setUpConsumer() throws JMSException {
        // 创建连接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        // 创建连接
        Connection connection = connectionFactory.createConnection();
        // 启动连接
        connection.start();
        // 创建Session
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列
        Destination adminQueue = session.createQueue(REQUEST_QUEUE);
        // 创建生产者
        producer = session.createProducer(null);
        // 设置持久化模式
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        // 创建消费者
        consumer = session.createConsumer(adminQueue);
        // 消费者设置消息监听
        consumer.setMessageListener(this);
    }

    public void stop() throws Exception {
        producer.close();
        consumer.close();
        session.close();
        brokerService.stop();
    }
    
    @Override
    public void onMessage(Message message) {
        try {
            // 创建新消息
            TextMessage response = this.session.createTextMessage();

            // 判断消息是否是文本消息
            if (message instanceof TextMessage) {
                // 强转为文本消息 
                TextMessage textMessage = (TextMessage) message;
                // 获取消息内容
                String text = textMessage.getText();
                // 设置消息
                response.setText(handleRequest(text));
            }
            response.setJMSCorrelationID(message.getJMSCorrelationID());
            producer.send(message.getJMSReplyTo(), response);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    /**
     * 构建消息内容
     * @param text 文本
     * @return
     */
    private String handleRequest(String text) {
        return "Response to '" + text + "'";
    }
    
    public static void main(String[] args) throws Exception {
        Server server = new Server();
        // 启动
        server.start();
        System.out.println();
        System.out.println("Press any key to stop the server");
        System.out.println();
        System.in.read();
        server.stop();
    }
}

III. Client代码

public class Client implements MessageListener {
    // 经纪人链接
    private static final String BROKER_URL = "tcp://localhost:61616";
    // 请求队列
    private static final String REQUEST_QUEUE = "requestQueue";
    // 连接
    private Connection connection;
    // 会话
    private Session session;
    // 生产者
    private MessageProducer producer;
    // 消费者
    private MessageConsumer consumer;
    // 请求队列
    private Queue tempDest;
    
    public void start() throws JMSException {
        // 连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        // 创建连接
        connection = activeMQConnectionFactory.createConnection();
        // 开启连接
        connection.start();
        // 创建会话
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列
        Destination adminQueue = session.createQueue(REQUEST_QUEUE);
        // 创建生产者
        producer = session.createProducer(adminQueue);
        // 设置持久化模式
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        // 创建模板队列
        tempDest = session.createTemporaryQueue();
        // 创建消费者
        consumer = session.createConsumer(tempDest);
        // 设置消息监听
        consumer.setMessageListener(this);      
    }
    
    /**
     * 停止
     * @throws JMSException 
     */
    public void stop() throws JMSException {
        producer.close();
        consumer.close();
        session.close();
    }
    
    /**
     * 请求
     * @param request
     * @throws JMSException 
     */
    public void request(String request) throws JMSException {
        System.out.println("Request: " + request);
        // 创建文本消息
        TextMessage textMessage = session.createTextMessage();
        // 设置文本内容
        textMessage.setText(request);
        // 设置回复
        textMessage.setJMSReplyTo(tempDest);
        // 获取UUID
        String correlationId = UUID.randomUUID().toString();
        // 设置JMS id
        textMessage.setJMSCorrelationID(correlationId);
        // 发送消息
        this.producer.send(textMessage);
    }

    @Override
    public void onMessage(Message message) {
        try {
            System.out.println("Received response for: " + ((TextMessage)message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) throws JMSException, InterruptedException {
        Client client = new Client();
        // 启动
        client.start();
        int i = 0;
        while(i++ < 10) {
            client.request("REQUEST- " + i);
        }
        Thread.sleep(3000);
        client.stop();
    }
}

IV. 测试

启动Server

图12.png
启动Client

图13.png

分布式--ActiveMQ 消息中间件(一)

1. ActiveMQ

1). ActiveMQ

2). Java Message Service(JMS)

3). JMS术语

4). ActiveMQ下载

5). 配置

6). 启动

7). 消息中间件(MOM：Message Orient middleware)

8).什么情况下使用ActiveMQ?

2. ActiveMQ 示例

1). P2P 示例

2). request/reply模型

代码下载

中间件

热门文章

最新文章

相关课程

相关电子书

相关实验场景