小白入门微服务(2) - 消息队列初体验

时间 2019-11-17

标签入门微服消息队列体验繁體版

原文原文链接

概述

前言
消息队列使用场景
什么是消息队列
经常使用消息队列库对比
Kafka 初体验
RabbitMQ 初体验
后记

前言

前面两篇咱们学习了node

接下来咱们来学习微服务中的异步通讯 - 消息队列。在这篇文章的学习中，默认你已经掌握了 Docker、docker-compose 的知识。若是你尚未掌握，能够翻阅个人历史文章。若是个人文章对你有帮助，欢迎关注、点赞、转发，这样我会更有动力作原创分享。python

消息队列的使用场景

异步处理

场景描述：在正常的用户注册流程中，用户注册完成都须要，发送邮件与短信，其中传统状况下，有串行与并行两种方式。docker

注册成功以后，写入数据库，写完以后，再发邮件，最后再发短信。整体耗时：

50 + 50 + 50 = 150 ms数据库

注册成功以后，写入数据库，并行发邮件和发短信。整体耗时：

50 + 50 = 100 msbootstrap

注册成功以后，写入数据库，将写入成功的信息，发送至消息队列，因为消费者本身去消息队列中取消息，这样在写入消息以后，便可成功返回。整体耗时：

50 + 5 = 55 msapi

三个数据一对比，谁优谁劣就很明显了bash

应用解耦

场景说明：用户下单后，须要减去库存系统中相应数量的库存。服务器

在传统模式下（上图），若是库存系统出现错误，则订单建立失败，并且二者过分耦合，对后面的新需求与维护也是至关大的挑战。在原有的架构上进行升级，则有下图：

订单系统：用户下单后，订单系统进行数据持久化处理，而后将消息写入消息队列，返回订单建立成功
库存系统：订阅消息，获取下单信息，库存系统根据订单信息，进行库存操做。

当下，二者就解耦了，库存系统出现错误，也能够正常进行下单了。由于只是入门文章，应用场景就先介绍到这里。更多使用场景请自行百度、Google。网络

什么是消息队列

如图，P 为 producer 生产者，C 为 consumer 消费者，红色部分为消息队列。通俗地解释一下消息队列，你想象一个场景：你到报社订阅了一份报纸，报社每日生产一份新报纸，便将新报纸发往邮局并告诉邮局你的地址，邮递员将你的报纸送往你的邮箱，你即可以愉快地阅读今天的时事新闻了。固然，可能一我的订阅了好几家报社，一家报社也能够被多我的订阅。在这个场景中，消息队列就担任了，邮箱、邮局、邮递员的角色。架构

经常使用消息队列库对比

在网络上搜索到一个比较全面的对比图，我这边就直接引用原图，再也不造轮子了。原图地址：（cloud.tencent.com/developer/a…

普遍来讲，电商、金融等对事务性要求很高的，能够考虑RabbitMQ和RocketMQ，对性能要求高的可考虑Kafka。

Kafka 初体验

kafka 术语：

Broker：Kafka 集群包含一个或多个服务器，这种服务器被称为 broker。
Topic：每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 Topic。（物理上不一样 Topic 的消息分开存储，逻辑上一个 Topic 的消息虽然保存于一个或多个 broker 上，但用户只需指定消息的 Topic 便可生产或消费数据而没必要关心数据存于何处）。
Partition：Partition 是物理上的概念，每一个 Topic 包含一个或多个 Partition。
Producer：负责发布消息到 Kafka broker。
Consumer：消息消费者，向 Kafka broker 读取消息的客户端。
Consumer Group：每一个 Consumer 属于一个特定的 Consumer Group（可为每一个 Consumer 指定 group name，若不指定 group name 则属于默认的 group）。

安装单节点 kafka

因为这里只是入门，就只使用单节点了。这里使用 Docker 来构建 kafka，其代码以下：docker-compose.yaml

version: '3'
services:
#<!--定义zk层服务-->
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
#<!--定义Kafka层-->
  kafka:
    image: wurstmeister/kafka
    depends_on: [ zookeeper ]
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 123.45.567.89
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
复制代码

docker-compose up 启动 kafaka。

nodejs 与 python 互调

咱们先来看看，nodejs 为 producer ，python 为 comsumer 的状况：

再看看，python 为 producer，nodejs 为consumer 的状况：

代码展现

nodejs producer

var kafka = require('kafka-node'),
    Producer = kafka.Producer,
    KeyedMessage = kafka.KeyedMessage,
    client = new kafka.KafkaClient({kafkaHost: '123.45.567.89:9092'}),
    producer = new Producer(client),
    payloads = [
        { topic: 'my_favorite_topic', messages: 'produce by nodejs', partition: 0 }
    ];
producer.on('ready', function () {
    producer.send(payloads, function (err, data) {
        console.log(data);
    });
});
producer.on('error', function (err) {})
复制代码

nodejs consumer

var kafka = require('kafka-node'),
    Consumer = kafka.Consumer,
    client = new kafka.KafkaClient({kafkaHost: '123.45.567.89:9092'}),
    consumer = new Consumer(
        client,
        [
            {topic: 'my_favorite_topic', partition: 0}
        ],
        {
            autoCommit: false
        }
    );
consumer.on('message', function (message) {
    console.log(message);
});
复制代码

python producer

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['123.45.567.89:9092'],
                         api_version=(0, 10, 1)
                         )  # 此处ip能够是多个['0.0.0.1:9092','0.0.0.2:9092','0.0.0.3:9092' ]
for _ in range(100):
    producer.send('my_favorite_topic', b'produce by python')
producer.close()
复制代码

python consumer

from kafka import KafkaConsumer
consumer = KafkaConsumer('my_favorite_topic', bootstrap_servers=['123.45.567.89:9092'], api_version=(0, 10, 1)
for msg in consumer:
    print(msg)
复制代码

RabbitMq 初体验

介绍过 Kafka 以后，RabbitMQ 就不在详细介绍了，请看源码了。

后记

不知不觉，消息队列也讲完了，要真真切切地区实践，才能真正地掌握。我的的知识储备老是有限的，若有错误的地方，还请大佬斧正。点击阅读原文，连接到个人知乎，我会在知乎上对文章错误的地方进行修改。

本篇文章首发于公众号「zone7」，关注公众号获取最新推文，后台回复【小白微服务】获取源码。