Apache RocketMQ 发布第一个 Python
客户端,该客户端基于CPP客户端的内核,通过使用boost
python封装C语言的API而实现,并由社区开发者贡献,

中间件

非底层操作系统软件,非业务应用软件,不能直接给最终用户使用和带来价值的软件。

图片 1服务调用的耦合.PNG图片 2利用消息中间件解耦服务.PNG

该版本提供以下特征:

消息中间件

关注于数据的发送和接收,利用高可靠的异步消息传递机制集成分布式系统。

此种方式,用户不需要等待短信服务和积分等其他服务处理完成才返回登录结果了。

  • 支持消息的可靠同步发送;

  • 支持可靠的推送消费模式;

  • 支持默认集群消费;

  • 支持延迟消息;

  • 支持自定义消息属性;

  • 支持消息压缩。

AMQP

AMQP(Advanced Message Queuing
Protocol)是一个提供统一消息服务的应用层标准协议,基于此协议的客户端和消息中间件可传递消息。,并不受客户端/中间件不同产品,不同开发语言等条件的限制。

①解耦②异步③横向扩展④安全可靠⑤顺序保证

(文/开源中国)    

JMS

Java消息服务(Java Message
Service)是一个Java平台中面向消息中间件的API,用于在两个应用程序间或分布式系统中发送消息,进行异步通信。

非底层操作系统软件,非业务应用软件,不是直接给最终用户使用的,不能直接给客户带来价值的软件统称为中间件。

JMS相关概念
  • 提供者:实现JMS规范的消息中间件服务器
  • 客户端:发送或接收消息的应用程序
  • 生产者/发布者:创建并发送消息的客户端
  • 消费者/订阅者:接收并处理消息的客户端
  • 消息:应用程序之间传递的数据内容
  • 消息模式:在客户端之间传递消息的方式,JMS中定义了主题和队列两种模式

关注于数据的发送和接受,利用高效可靠地异步消息传递机制集成分布式系统

JMS消息模式

图片 3消息中间件.PNG

队列模型
  • 客户端包括生产者和消费者
  • 队列中的每个消息只能被一个消费者消费
  • 消费者可以随时消费队列中的消息

Java消息服务(Java Message
Service)即JMS,是一个Java平台中关于面向消息中间件的API,用于在两个应用程序之间,或在分布式系统中发送消息,进行异步通信。

主题模型
  • 客户端包括发布者和订阅者
  • 主题中的消息被所有订阅者消费
  • 消费者不能消费订阅之前就发送到主题中的信息

AMQP(Advanced Message Queuing
Protocol)是一个提供统一消息服务的应用层标准高级消息队列协议,基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同的开发语言等条件的限制。

JMS编码接口
  • ConnectionFactory 用于创建连接到消息中间件的连接工厂
  • Connection 代表了应用程序和消息服务器之间的通信链路
  • Destination 指消息发布和接收的地点,包括队列和主题
  • Session 表示一个单线程的上下文,用于发送和接收消息
  • MessageCosumer 由会话创建,用于接收发送到目标的消息
  • MessageProducer 由会话创建,用于发送消息到目标
  • Message
    是在消费者和生产者之间传送的对象,消息头,一组消息属性,一个消息体

图片 4JMS与AMQP对比.PNG

生产者实现
/**
 * 生产者
 */
public class QueueProducer {

    // ActiveMQ地址
    private static final String URL = "tcp://127.0.0.1:61616";

    // 队列名称
    private static final String QUEUE_NAME = "queue.test";

    private static final Logger log = LoggerFactory.getLogger(QueueProducer.class);

    public static void main(String[] args) throws JMSException {

        // 创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);

        // 创建Connection
        Connection connection = connectionFactory.createConnection();

        // 启动连接
        connection.start();

        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 创建Destination 
        Destination destination = session.createQueue(QUEUE_NAME);

        // 创建生产者
        MessageProducer producer = session.createProducer(destination);

        for (int i = 0; i < 100; i++) {

            // 创建消息
            TextMessage textMessage = session.createTextMessage("test: " + i);

            // 发布消息
            producer.send(textMessage);

            log.info("send message: " + textMessage.getText());
        }

        // 关闭连接
        connection.close();
    }
}

ActiveMQ是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ是一个完全支持JMS1.1和J2EE1.4规范的JMS
Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今J2EE应用中间件仍然扮演者特殊的地位。

消费者实现
/**
 * 消费者
 */
public class QueueConsumer {

    // ActiveMQ地址
    private static final String URL = "tcp://127.0.0.1:61616";

    // 队列名称
    private static final String QUEUE_NAME = "queue.test";

    private static final Logger log = LoggerFactory.getLogger(QueueConsumer.class);

    public static void main(String[] args) throws JMSException {

        // 创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);

        // 创建Connection
        Connection connection = connectionFactory.createConnection();

        // 启动连接
        connection.start();

        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 创建目标
        Destination destination = session.createQueue(QUEUE_NAME);

        // 创建消费者
        MessageConsumer consumer = session.createConsumer(destination);

        // 创建监听器,异步监听消息并处理(消费)
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    log.info("receive message: " + textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

ActiveMQ特性:

ActiveMQ

  • 多种语言和协议编写客户端。语言:Java、C、C++、C#、Ruby、Perl、Python、PHP应用协议:OpenWire、Stomp、REST、WS
    Notification、XMPP、AMQP
  • 完全支持JMS1.1和J2EE1.4规范(持久化、XA消息、事务)
  • 虚拟主题、组合目的、镜像队列
安装
  • Windows:解压安装包后通过bin目录下的activemq.bat启动服务,或使用InstallService.bat安装服务并启动
  • Linux 解压安装包后进入bin目录,通过./activemq start和./activemq
    stop启动和停止服务

RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

控制台

访问ActiveMQ控制台:http://127.0.0.1:8161/admin/(默认登录账号/密码:admin/admin)。

在导航栏可以访问Queues(队列消息)和Topics(主题消息)。

图片 5

ActiveMQ控制台导航栏

通过控制台可以看到队列名、待消费消息的数量、当前消费者数量、入队消息数量、出队消息数量和操作等。

图片 6

image.png

RabbitMQ特性:

集群
  • 实现高可用,以排除单点故障引起的服务中断
  • 实现负载均衡,以提升效率为更多客户提供服务
  • 支持多种客户端如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript等
  • AMQP的完整实现(vhost、Exchange、Binding、Routing Key等)
  • 事务支持/发布确认
  • 消息持久化
集群方式
  • 客户端集群:让多个消费者消费同一个队列
  • Broker clusters:多个Broker之间同步消息
  • Master/Slave:实现高可用

Kafka是一种高吞吐量的分布式发布订阅消息系统,是一个分布式、分区的、可高的分布式日志存储服务。它通过一种独一无二的设计提供了一个消息系统的功能。

Spring对JMS的支持

Kafka特性:

JmsTemplate

用于发送消息

  • 在Spring容器中注册JmsTemplate即可操作JMS
  • JmsTemplate是线程安全的,可以在整个应用内操作
  • 通过O的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能
  • 高吞吐量:即使是非常普通的硬件Kafka也可以支持每秒数百万的消息
  • Partition、Consumer Group
DefaultMessageListenerContainer

消息监听容器

  • 配置JMS连接
  • 配置消费队列
  • 配置消费监听器
  • 配置消费者数量

图片 7各个消息中间件对比.PNG

依赖
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-core</artifactId>
        <version>5.7.0</version>
    </dependency>

Java消息服务(Java Message
Service)应用程序接口,是一个Java平台中关于面向消息中间件的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。

spring-jms配置文件

配置生产者、Destination和消费者

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd">

    <!-- 配置扫描jms包 -->
    <context:component-scan base-package="com.wch.ssm.jms"/>

    <!-- ActiveMQ提供的ConnectionFactory,符合JMS规范 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://127.0.0.1:61616"/>
    </bean>

    <!-- Spring jms提供的连接池 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>

    <!-- 队列Destination -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="queue.test"/>
    </bean>

    <!-- 主题Destination -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="topic.test"/>
    </bean>

    <!-- 注册JmsTemplate,用于发送消息 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>

    <!-- 配置消息监听器 -->
    <bean id="consumerTextMessageListener" class="com.wch.ssm.jms.consumer.ConsumerTextMessageListener"/>

    <!-- queueDestination消息监听容器 -->
    <bean id="jmsContainer1" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="queueDestination"/>
        <property name="messageListener" ref="consumerTextMessageListener"/>
        <!-- 配置多个消费者 -->
        <property name="concurrency" value="3-5"/>
    </bean>

    <!-- topicDestination消息监听容器 -->
    <bean id="jmsContainer2" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="topicDestination"/>
        <property name="messageListener" ref="consumerTextMessageListener"/>
    </bean>

</beans>
  • 提供者:实现JMS规范的消息中间件服务器
  • 客户端:发送或接收消息的应用程序
  • 生产者/发布者:创建并发送消息的客户端
  • 消费者/订阅者:接收并处理消息的客户端
  • 消息:应用程序之间传递的数据内容
  • 消息模式:在客户端之间传递消息的模式,JMS中定义了主题和队列两种模式
配置生产者及其实现类
public interface ProducerService {

    /**
     * 发送TextMessage
     *
     * @param message TextMessage
     */
    void sendTextMessage(String message);
}

@Service
public class ProducerServiceImpl implements ProducerService {

    @Resource
    private JmsTemplate jmsTemplate;

    @Resource(name = "queueDestination")
    private Destination queueDestination;

    @Resource(name = "topicDestination")
    private Destination topicDestination;

    private static final Logger log = LoggerFactory.getLogger(ProducerServiceImpl.class);

    @Override
    public void sendTextMessage(String message) {
        // 创建MessageCreator
        MessageCreator messageCreator = new MessageCreator() {
            /**
             * 创建TextMessage
             * @param session session
             * @return TextMessage
             * @throws JMSException JMSException
             */
            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage textMessage = session.createTextMessage(message);
                log.info("{} produce message: {}", Thread.currentThread().getName(), textMessage.getText());
                return textMessage;
            }
        };

        for (int i = 0; i < 1000; i++) {

            // 发送队列消息
            jmsTemplate.send(queueDestination, messageCreator);

            // 发送主题消息
            jmsTemplate.send(topicDestination, messageCreator);
        }
    }
}

配置消费者

要求实现MessageListener接口,对监听到的消息进行处理(如何消费)

public class ConsumerTextMessageListener implements MessageListener {

    private static final Logger log = LoggerFactory.getLogger(ConsumerTextMessageListener.class);

    @Override
    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage) message;
        try {
            log.info("{} consume message: {}", Thread.currentThread().getName(), textMessage.getText());
        } catch (JMSException e) {
            log.error("consume message failed");
        }
    }
}
  • 客户端包括生产者和消费者
  • 队列中的消息只能被一个消费者消费
  • 消费者可以随时消费队列中的消息
测试
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:spring/spring-jms.xml")
public class JmsTest {

    @Resource
    private ProducerService producerService;

    @Test
    public void jmsTest() {
        producerService.sendTextMessage("test");
    }
}

队列模型示意图:

控制台打印

图片 8

控制台打印

图片 9队列模型示意图.PNG

SpringBoot集成JMS

  • 客户端包括发布者和订阅者
  • 主题中的消息被所有订阅者消费
  • 消费者不能消费订阅之前就发送到主题中的消息
依赖
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>

主题模型示意图:

配置

SpringBoot不能同时支持queue和topic两种消息模型,默认支持queue,通过配置特定的ListenerContainer使得同时支持两种消息模型。

@Configuration
public class JmsConfig {

    @Value("${jms-name.queue-test}")
    private String queueName;

    @Value("${jms-name.topic-test}")
    private String topicName;

    /**
     * 配置Queue
     * @return Queue
     */
    @Bean
    public Queue queue() {
        return new ActiveMQQueue(queueName);
    }

    /**
     * 配置Topic
     * @return Topic
     */
    @Bean
    public Topic topic() {
        return new ActiveMQTopic(topicName);
    }

    /**
     * 配置JmsMessagingTemplate,对JmsTemplate的封装
     * @param jmsTemplate jmsTemplate
     * @return JmsMessagingTemplate
     */
    @Bean
    public JmsMessagingTemplate jmsMessagingTemplate(JmsTemplate jmsTemplate) {
        return new JmsMessagingTemplate(jmsTemplate);
    }

    /**
     * 为Topic配置MessageListenerContainer,默认为Queue Container
     * @param connectionFactory connectionFactory
     * @return JmsListenerContainerFactory
     */
    @Bean
    public JmsListenerContainerFactory<?> topicListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        // 设置topic消息类型
        factory.setPubSubDomain(true);
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
}

图片 10主题模型示意图.PNG

配置生产者
@Service
@EnableScheduling
public class Producer {

    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    @Resource
    private Queue queue;

    @Resource
    private Topic topic;

    private static int count = 0;

    private static final Logger log = LoggerFactory.getLogger(Producer.class);

    /**
     * 每1000ms分别生产一条消息给两个消息模型
     */
    @Scheduled(fixedRate = 1000)
    public void sendTextMessage() {

        String queueMessage = "to queue " + count;

        String topicMessage = "to topic " + count++;

        // 发送消息给队列
        log.info("{} produce message: {}", Thread.currentThread().getName(), queueMessage);
        jmsMessagingTemplate.convertAndSend(this.queue, queueMessage);

        // 发送消息给主题
        log.info("{} produce message: {}", Thread.currentThread().getName(), topicMessage);
        jmsMessagingTemplate.convertAndSend(this.topic, topicMessage);
    }
}
  • ConnectionFactory:用于创建连接到消息中间件的连接工厂
  • Connection:代表了应用程序和消息服务器之间的通信链路
  • Destination:指消息发布和接收的地点,包括队列和主题
  • Session:表示一个单线程的上下文,用于发送和接收消息
  • MessageConsumer:由会话创建,用户接收发送到目标的消息
  • MessageProducer:由会话创建,用于发送消息到目标
  • Message:是在消费者和生产者之间传送的对象,消息头,一组消息属性,一个消息体
配置消费者
@Service
public class Consumer {

    private static final Logger log = LoggerFactory.getLogger(Consumer.class);

    // 设置监听多个消息模型
    @JmsListeners(value = {
            // 监听queue,设置1-3个消费者
            @JmsListener(destination = "${jms-name.queue-test}", concurrency = "1-3"),
            // 监听topic,配置特定的监听容器
            @JmsListener(destination = "${jms-name.topic-test}", containerFactory = "topicListenerContainerFactory")
    })
    public void consumeMessage(Message message) throws JMSException {
        TextMessage textMessage = (TextMessage) message;
        log.info("{} consume message: {}", Thread.currentThread().getName(), textMessage.getText());
    }
}

JMS编码接口之间的关系:

图片 11JMS编码接口之间的关系.PNG