扫盲消息队列 | 消息中间件 | Kafka

王炸

发布于 2020-04-26 14:28:17

1.9K00

代码可运行

文章被收录于专栏：转行程序员转行程序员

运行总次数：0

代码可运行

背景

分布式微服务系统下，凡是可以“排队”去做的事情，都可以使用消息队列。网上买东西同样也需要“排队付款”，但是有人说，我点确认付款后马上就显示成功了，没感觉到排队呀？其实在后台系统中是排了，只不过排队的时间对于人来说有点短，可能1-2秒就结束了，但是对于计算机来说，这1-2秒的时间很长了。

大型分布式系统建设中，消息队列主要解决应用耦合、异步消息、流量削锋等问题。实现高性能、高可用、可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。

Web应用程序毫无疑问有大量的代码执行HTTP请求/响应周期的一部分。这适用于更快的任务耗费数百毫秒内或更少。然而,有些处理,还需要耗时更多甚至最终会是一两秒钟缓慢的同步执行，在如此长时间的调用流转中，肯定有一些调用是可以不同步的，如下单送积分，用户下单是最主要的，送积分的操作可以异步去做，订单支付成功给用户的短信通知，返回支付订单进入下一环节更更好，短信通知可以异步去发送，为了应对诸如此类的异步操作，消息队列这门技术应运而生。

message-queue-example

在计算机科学中，消息队列（Message queue）是一种进程间通信或同一进程的不同线程间的通信方式。实际上，消息队列常常保存在链表结构中。[2]拥有权限的进程可以向消息队列中写入或读取消息。

目前，有很多消息队列有很多开源的实现，包括JBoss Messaging、JORAM、Apache ActiveMQ、Sun Open Message Queue、RabbitMQ[3]、IBM MQ[4]、Apache Qpid[5]、Apache RocketMQ[6]和HTTPSQS。[7]

说了这么多没用的，消息队列到底在企业里怎么用的？

我就直接上两段代码吧

发送一条消息demo

public class MqProducer {
    private final Logger LOG = LoggerFactory.getLogger(MqProducer.class);

    @Resource
    private Producer payProducer;

    public void sendPayMsg(String msg) {
        try {
            LOG.debug("send msg:{}", msg);
            payProducer.send(msg);//发送出去一条消息。
        } catch (MQException e) {
            LOG.error("mq消息异常 message:{}", msg, e);
        }
    }
}

接收一个消息demo

public class DemoConsumer {

    /**
    * 注意：服务端对单ip创建相同主题相同队列的消费者实例数有限制，超过100个拒绝创建.
    * */
    private static IConsumerProcessor consumer;

    public static void main(String[] args) throws Exception {
        Properties properties = new Properties();
            properties.setProperty(ConsumerConstants.SubscribeGroup, "dache.risk.log.queue.v2");
 
        // 创建topic对应的consumer对象（注意每次build调用会产生一个新的实例）
        consumer = KafkaClient.buildConsumerFactory(properties, "topic.xxx.xxx");
 
        // 调用recvMessageWithParallel设置listener
        consumer.recvMessageWithParallel(String.class, new IMessageListener() {
            @Override
            public ConsumeStatus recvMessage(Message message, MessagetContext context) {
                //TODO:业务侧的消费逻辑代码
                try {
                    System.out.println("message=[" + message.getBody() + "]  partition=" + message.getParttion());
                } catch (Exception e) {
                    e.printStackTrace();
                }
              
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
       
    }
}

消息长什么样子？

{"businessType":1,"cityId":10,"ctime":1567426767077,"dataKey":20190902,"logType":1,"phone":"13212341234","uid":12345678,"userType":1,"uuid":"32EA02C86D78863"}

代码呢，就是普通的java代码，只不过引入了一个kafka的jar，消息就是json串，使用消息队列真的就这么点代码，剩下的内容都是业务代码了。

新手关注消息队列，主要关注两个最重要的概念就行了，一个是生产者，一个是消费者，两者的关系和我们日常发短信一样，短信是通过手机号发送接收，系统间消息是通过topic，可以理解成手机号。

Producer消息的生产方，如支付系统确认用户已经支付，支付系统要通知订单系统和物流系统，支付系统就是生产者。
Consumer消费的接收方，Producer 的案例中，物流系统就是消费方，前两个都比较简单，我就不多说了。
Topic每条发布到MQ集群的消息都有一个类别，这个类别被称为topic，可以理解成一类消息的名字。所有的消息都以topic作为单位进行归类。

日常开发中需要关心哪些指标

1.生产消息数目

每分钟几百几千个都正常水平吧，业务繁忙的每分钟几万几十万也是有的

image-20200420193350371

2.消息延迟情况

延迟越低越好啦，几百毫秒正常水平。

image-20200420193540766

3.消息积压数

这个当然是要0了，如果遇到消费端服务器上线，可能会有段时间积压正常，这个指标，日常应该都是0才对。

image-20200420194052328

为什么使用消息队列

开头不是说了，排队能解决一个问题，就是削峰，意思就是流量洪峰来了，收银员结账速度依旧是一样的，不会被累死，还有两个重要的概念就是解耦、异步

使用消息队列有什么缺点呢？

这个新手也一定要知道啦，因为面试官会问。

消息丢失问题： 任何系统不能保证万无一失，比如 Producer 发出了10000条消息，Consumer 只收到了 9999 个消息，有1个丢了，Consumer 能否接受丢一条？如果是订单成功短信可以接受丢一条，就是有一个顾客没有通知到已经发货，但货还是发出去了，如果是支付系统，用户已经付款却因为消息丢失没有通知到订单或物流系统，那恐怕顾客要找你麻烦了。
消息重复问题：如 Producer 发出了10000条消息，Consumer 只收到了 10001 条消息，有一条是重复的，业务能否接受一条重复的消息，这个是作为系统设计者要考虑的问题。
消息的顺序问题：如 Producer 发送顺序是123，Consumer 收到的消息是132，要考虑消费端是否对顺序敏感。
一致性问题： 如消息丢失问题真的发生且无法找回，会造成两个系统的数据最终不一致，如果消息延迟，会造成短暂不一致。