消息队列以及非常牛的kafka

Posted 2023-03-31 三思守心

目录

概念

例子比较

传统串行服务

引入消息队列

事务

socket

MQ的两种流派

1、有broker

2、无broker

kafka使用场景

kafka的基本概念

 kafka的安装

关于消息队列的思考：

---

 

1、对计算机有初步了解的人都知道，队列是个什么玩意，它就像一个管道一样先进先出。

那么我们来了解一下消息队列是个啥玩意吧！！！

概念

2、Message Query（MQ），消息队列中间件，很多初学者认为，MQ通过消息的发送和接受来实现程序的异步和解耦，mq主要用于异步操作，这个不是mq的真正目的，只不过是mq的应用，mq真正的目的是为了通讯。他屏蔽了复杂的通讯协议，像常用的dubbo，http协议都是同步的。这两种协议很难实现双端通讯，A调用B,B也可以主动调用A，而且不支持长连接。mq做的就是在这些协议上构建一个简单协议——生产者、消费者模型，mq带给我们的不是底层的通讯协议，而是更高层次的通讯模型。他定义了两个对象：发送数据的叫做生产者，接受消息的叫做消费者，我们可以无视底层的通讯协议，我们可以自己定义生产者消费者。

例子比较

我们以淘宝订单为例，比较传统串行服务和引入消息队列多的区别

传统串行服务

 如果我们想买一个东西，那么需要经过以上的几个步骤，完成之后，才会给我们发送成功信号，消耗的时间是长的，并且十分容易引发错误，当网络波荡的时候，就会导致任务失败。

传统串行服务的优点：系统架构简单，排查错误比较容易

传统串行服务的缺点：1.耦合性太强 2.系统吞吐量不大，耗时多

引入消息队列

 引入消息队列后，当在处理新订单是，只要消息发送到消息队列上，我们客户端就能收到订单成功的消息。剩下的处理操作，由消息队列来处理。大大减少了耗时。并且后面的操作之间也不会相互影响。因为他们进行的异步操作。这就降低了耦合性，提高了性能。

这里我们要思考一个问题：

如何解决重复通知，一个服务订阅多个队列的问题？

使用分布式锁，当一个消息队列被一个服务订阅以后，就会给消息队列加上锁，其他的服务就不能在订阅了。

事务

只要涉及到消息队列，肯定要处理事务问题。

事务：原子性，我所有的动作要么都执行，要么都不执行。

比如说：张三给我转100元，可以分成两部分，张三账户减100，我账户加100

socket

1、socket用于即时通讯。socket是长连接。

2、传统的http请求，随着服务端的返回给客户端消息，两者之间的关联就消失了，所以服务端想要主动联系客户端就得靠轮询的方式，但是这种轮询比较耗费性能的，既要消耗客户端性能，又消耗服务器资源，于是有了长连接的概念。

MQ的两种流派

1、有broker

broker是一个中转站，为了将消息放到不同的队列里。存放方式有两种：

（1）消息来了主动放

（2）等待消息请求

有broker的也分为两类

       1、重topic

        kafka：全球消息处理性能最快的一款mq

        rocket：阿里内部的一个大神根据kafka的执行原理手写的，性能与kafka差不多，但是功能   上比kafka要多，比如说顺序消费。

        2、轻topic       

        可以没有topic，topic只是一种中转模式

        rabbitmq

2、无broker

        zeromq：没有使用broker，是直接使用socket进行通信。

kafka使用场景

1，日志收集：可以用kafka收集各种服务的日志 ，通过已统一接口的形式开放给各种消费者。

2，消息系统：解耦生产和消费者，缓存消息。

3，用户活动追踪：kafka可以记录webapp或app用户的各种活动，如浏览网页，点击等活动，这些活动可以发送到kafka，然后订阅者通过订阅这些消息来做监控。

4，运营指标：可以用于监控各种数据。

kafka的基本概念

kafka是一个分布式的分区的消息，提供消息系统应该具备的功能。

名称	解释
broker	消息中间件处理节点，一个broker就是一个kafka节点，多个broker构成一个kafka集群。
topic	kafka根据消息进行分类，发布到kafka的每个消息都有一个对应的topic
producer	消息生产（发布）者
consumer	消息消费（订阅）者
consumergroup	消息订阅集群，一个消息可以被多个consumergroup消费，但是一个consumergroup只有一个consumer可以消费消息。

 kafka的安装

可以来看这篇博客

http://t.csdn.cn/LYJ3b

关于消息队列的思考：

了解了消息队列，现在让我们来思考一个消息队列的问题，也是经典的消息堆积问题，比如说客户端每秒生产10000条消息，但是消费者每秒消耗100条。

-------------------------------------------------------------

|

|

|
|

|

答案：

1、增加消费者，既然消费不了，那么就多找几个消费者，来完成消费

2、限流，限制客户端每秒产生消息的数量。

消息队列的作用以及kafka和activemq的对比

背景分析

消息队列这个类型的组件一直是非常重要的组件，当经过两家企业后我就很坚信这个结论了。队列这种东西，最广泛的作用还是在于解耦，宽泛一点的说，它可以将不同部门的工作内容进行有效的整合，基于一个约定好的格式，就可以两头互相不干扰的进行开发。可以说这个生产消费的思想不仅仅适用于程序也适用于非常多的地方。
目前对于我看到的来说，kafka更多的还是做为一个数据源，数据桥梁的作用，不同业务之间的沟通。比如需要实时接入A部门的业务数据的话，就会有这样的手段：

 

 

 

落地到HDFS的数据会用来进行一些算法上的离线处理，而kafka端则是给需要实时性的消费方。其实数据的消费方式无非也就实时和离线两种方式。

Kafka和activemq对比

相比过去经常使用的activemq，kafka确实非常的不同，做一个对比来深化印象

对比	Activemq	Kafka
接口协议	遵守JMS规范,各语言支持较好	没有遵循标准MQ接口协议,使用较为复杂
吞吐量	较低，磁盘随机读写	较高，磁盘顺序读写
游标位置	AMQ来管理，无法读取历史数据	客户端自己管理，不乐意甚至重新读一遍也行
HA机制	主从复制,竞争锁的方式来选举新的主节点	和hadoop系列产品一样,由zk管理所有节点

说到底，主要还是做为kafka的消费方，能感受到最大的不同还是在于几个：

1. 吞吐量确实非常高
   2.可以重读历史数据
   但是也有一些缺点：
   1.概念上比较复杂，相对于AMQ只需要知道ip和队列名你就能获得数据，Kafka使用起来非常繁琐

Kafka的基本概念（摘录）

1.Broker：消息中间件处理结点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。
2.Topic：一类消息，例如page view日志、click日志等都可以以topic的形式存在，Kafka集群能够同时负责多个topic的分发。
3.Partition：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。
4.Segment：partition物理上由多个segment组成。
5.offset：每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息.

Kafka消费端的常用参数

        Properties props = new Properties();
                //zk服务器的地址  xxxx:2181
        props.put("zookeeper.connect", zookeeper);
                //组的名称，区别于其他group否则会接收不到数据
        props.put("group.id", groupId);
        props.put("zookeeper.session.timeout.ms", "8000");
        props.put("zookeeper.connection.timeout.ms", "20000");
        props.put("zookeeper.sync.time.ms", "2000");
        props.put("auto.commit.interval.ms", "5000");
        props.put("rebalance.max.retries", "5");
        props.put("rebalance.backoff.ms", "60000");
        props.put("auto.commit.enable", "true");
                //重点参数，是否每次都从offset最前面开始读起
        props.put("auto.offset.reset", "smallest");

Kafka的一些常用命令

查看所有的topic

bin/kafka-topics.sh --zookeeper zk1.test-inf-zk.data.m.com:2181/octopus,zk2.test-inf-zk.data.m.com:2181/octopus,zk3.test-inf-zk.data.m.com:2181/octopus --list

查看topic的偏移量

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic xiuxiu_sync_search_big_data --time -1 --broker-list 192.168.199.11:9092 --partitions 0

查看topic的状态

bin/kafka-topics.sh --zookeeper 192.168.199.11:2181 --topic xiuxiu_sync_search_big_data --descr