Wicky's Blog

Kafka MQ的一些想法

发表于 | 分类于 |

背景

在公司系统架构发展中，有若干场景需要可靠的队列服务来出传递数据或消息，这些场景可能有:

  • 系统的数据同步，如保存在redis中的数据，需要落地到Mysql中，为了保证性能可扩展性，通过消息队列来执行数据的落地操作。
  • 跨系统间的数据同步，如内容仓库中新增、修改、删除操作导致各个系统间数据的变化
  • 事件的异步处理，如一些统计相关的数据

综合上面的一些场景，消息队列可以抽象成下面这样：

  • 高可用，消息队列服务应该保持尽可能高的可用性，尤其是写消息的操作
  • 低延迟，单次写队列的操作在1ms内完成，以提高吞吐量
  • 可扩展，当多个业务接入时，随着消息量的增长，应当保持相对容易的扩展并且对上下游的透明的。
  • 消息的安全性，保证消息不丢失并且可以回归(回源时间可控，一周以内)
  • 接入成本低，对语言无要求(通过HTTP协议)

系统设计

RedisQueue

  • supervise脚本，重启机器或脚本要重新reload
  • redis保存内存，没有持久化到磁盘，断电可能导致丢数据，可靠性差
  • pub/sub ,多写,一个操作通知多个业务方，上下游耦合严重，重复数据推到多个redis queue中
  • …..

KafkaQueue

logo
说明：

  • MqProxy

各上游子系统（如订单系统)，通过统一的MqProxy来提交消息。

MqProxy提供HTTP接口，上游提交消息时，声明Topic和消息内容。

MqProxy和Zookeeper保持长连接，以加速（实测PHP访问Zk大概有10-20ms的耗时）。

MqProxy处理消息的分区，在提交消息给MqProxy时，同时提供Partition Key。

  • Producer

MqProxy作为Producer将消息提交到Kafka。

  • Broker

参考Kafka的系统架构，Broker部分提供对某个Topic的Partation和Replica机制，即每一个Topic的数据，都可按某个主键进行分片，并支持多个副本,Replication保持数据备份。

Broker需要实现Leader选举机制，并通过和Zookeeper保存各Topic/Partation的Leader。

Producer和Broker交互之前，通过Zookeeper获得对应Leader，而后进行交互。

Broker支持可扩展。

  • Consumer

Consumer可支持队列服务本身的Consumer机制(Pull Offset)。

然而，现在的系统，大部分是PHP这样的系统，并不是通过deamon方式运行的，因此我们需要实现一个Pusher机制，使得消息的消费从拉模型变为推模型，使得下游需要接受消息的子系统，更低成本的接入。

因此，我们实现一套Push Consumer机制。

  • Pusher Consumer

Pusher Consumer是可配置的常驻服务，可指定Topic->url，支持多个。

配置文件举例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
{
    "consumer_groups": [                //support multi callback url
        {
            "worker_num": 16,
            "url": "http://localhost",  //callback url
            "retry_times": 4,           //will retry if the callback request response non 200 code
            "bypass_failed":true,       //auto jump to next if single message were processed failed,set to false if service need to process message exactlly
            "failed_sleep":"5s",       //when bypass_failed set to true,sleep this time before retry
            "timeout": "3s",            //the callback time
            "topics": [                 //topic(s) consumed
                "test"
            ],
            "zookeepers": [             //zookeeper hosts
                "127.0.0.1:2181"
            ],
        }
    ]
}

pusher consumer会将消息有序推送到下游系统，各个下游系统都能够拿到所有的消息，自行进行处理，且进度各自独立，互不影响。

pusher consumer通过zookeeper保存各个下游当前的命令点，因此也是无状态的。

  • http callback

下游接受消息的接口，提供统一的消息接受格式。

同时，由于Pusher Consumer采用at least once的方式（即消息至少传输一次，由于网络等原因导致的超时或出错，pusher会自动重发直到超过retry次数或者返回成功），建议下游的接口是幂等，即多次调用也不会使得状态错误。