消息中间件解决方案JMS(1)-CFANZ编程社区

1.什么是消息中间件

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。对于消息中间件，常见的角色大致也就有Producer（生产者）、Consumer（消费者）,主要作用是改善系统模块调用关系、减少模块之间的耦合。

常见的消息中间件产品:

（1）ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。我们在本次课程中介绍 ActiveMQ的使用。

（2）RabbitMQ

AMQP协议的领导实现，支持多种场景。淘宝的MySQL集群内部有使用它进行通讯，OpenStack开源云平台的通信组件，最先在金融行业得到运用。

（3）ZeroMQ

史上最快的消息队列系统

（4）Kafka

Apache下的一个子项目。特点：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统。适合处理海量数据。

2.什么是JMS

JMS（Java Messaging Service）是Java平台上有关面向消息中间件的技术规范，它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。

JMS本身只定义了一系列的接口规范，是一种与厂商无关的 API，用来访问消息收发系统。它类似于 JDBC(java Database Connectivity)：这里，JDBC 是可以用来访问许多不同关系数据库的 API，而 JMS 则提供同样与厂商无关的访问方法，以访问消息收发服务。许多厂商目前都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，这只是几个例子。 JMS 使您能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个 JML 客户机发送消息。消息是 JMS 中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。

JMS 定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一

些不同形式的数据，提供现有消息格式的一些级别的兼容性。

· TextMessage--一个字符串对象

· MapMessage--一套名称-值对

· ObjectMessage--一个序列化的 Java 对象

· BytesMessage--一个字节的数据流

· StreamMessage -- Java 原始值的数据流

3.JMS消息传递类型

对于消息的传递有两种类型：

一种是点对点的，即一个生产者和一个消费者一一对应；

另一种是发布/ 订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

4.SpringBoot 集成 ActiveMQ

（1）创建一个springboot项目，添加依赖

（2）application.properties文件的配置

（3）启动类增加 @EnableJms 注解

（4）初始化和配置 ActiveMQ 的连接

package com.weiwei.activemq.config;


import org.apache.activemq.ActiveMQConnectionFactory;

import org.apache.activemq.command.ActiveMQQueue;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.jms.config.JmsListenerContainerFactory;

import org.springframework.jms.config.SimpleJmsListenerContainerFactory;

import org.springframework.jms.core.JmsMessagingTemplate;



import javax.jms.ConnectionFactory;

import javax.jms.Queue;



@Configuration

public class ActivemqConfig {

    @Value("${spring.activemq.broker-url}")

    private String brokerUrl;

    @Value("${spring.activemq.user}")

    private String username;

    @Value("${spring.activemq.password}")

    private String password;

    @Value("${spring.activemq.queue-name}")

    private String queueName;



    //创建链接工厂

    @Bean

    public ConnectionFactory connectionFactory(){

        return new ActiveMQConnectionFactory(username,password,brokerUrl);

    }

    //jms的消息模板

    @Bean

    public JmsMessagingTemplate jmsMessagingTemplate(){

        return new JmsMessagingTemplate(connectionFactory());

    }

    //点对点消息对象

    @Bean(name = "queue")

    public Queue queue(){

        return new ActiveMQQueue(queueName);

    }

    //在Queue模式中，对消息的监听,需要对containerFactory进行配置

    @Bean("queueListner")

    public JmsListenerContainerFactory<?> jmsListenerContainerFactory(){

        SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();

        factory.setConnectionFactory(connectionFactory());

        factory.setPubSubDomain(false);

        return factory;

    }



}

（5）创建点对点消息的发送者

package com.weiwei.activemq.controller;



import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.jms.core.JmsMessagingTemplate;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;



import javax.jms.Queue;



/**

 * 点对点模式消息发送者

 */

@RestController

public class QueueProController {

    //jms消息模板

    @Autowired

    private JmsMessagingTemplate jmsMessagingTemplate;

    //点对点消息对象

    @Autowired

    private Queue queue;



    //消息发送方法

    @RequestMapping("/sendqueue.do")

    public String sendQueue(String username){

        jmsMessagingTemplate.convertAndSend(this.queue,username);

        return "send-success-queue";

    }

}

（6）创建点对点消息类型的接收者

/**

 * 点对点的消息接受者

 */
package com.weiwei.activemq.controller;



import org.springframework.jms.annotation.JmsListener;

import org.springframework.stereotype.Component;



@Component

public class QueueConsumerListener {

    //queue模式的消费者

    @JmsListener(destination = "${spring.activemq.queue-name}",containerFactory = "queueListener")

    public void readActiveQueue(String message){

        System.out.println("queue接收到："+message);

    }

}

（7）在主类中启动消息服务，在浏览器的地址栏输入

http://127.0.0.1:8080/sendqueue.do?username=zhangsan

控制台可以看到具体的消息内容--------------queue接收到：zhangsan