1000Wqps生产级IM，怎么架构？

▌前言：

在40岁老架构师尼恩的读者社区(50+)中，很多小伙伴拿高薪，完成架构的升级，进入架构师赛道，打开薪酬天花板。

然后，在架构师的面试过程中，常常会遇到IM架构的问题：

如果要你从0到1做IM架构，需要从哪些方面展开

你是怎么做项目的IM架构的？

1亿级以上qps的高并发IM，该如何架构？

前几天，40岁老架构师尼恩，站在腾讯企业IM的巨人肩膀，给大家提供一份比较全面的参考答案。具体的文章链接为

腾讯太狠：10亿QPS的IM，如何实现？

今天，40岁老架构师尼恩，站在B站1000Wqps生产级IM服务框架的巨人肩膀，再给大家提供一份比较全面的参考答案。就是本文。

通过这些企业级、工业级、生产级案例，大家可以在面试的时候，对比进行介绍，综合介绍。

从而给面试官展示自己雄厚的技术实力、开阔的技术视野，从而在面试的时候，可以充分展示一下大家雄厚的 “技术肌肉”，让你的面试官爱到 “不能自已、口水直流”。

这里，尼恩也一并把这个题目以及参考答案，收入咱们的《尼恩Java面试宝典》V92版本，供后面的小伙伴参考，提升大家的 3高架构、设计、开发水平。

注：本文以 PDF 持续更新，最新尼恩架构笔记、面试题的PDF文件，请到公号【技术自由圈】取

▌本文目录：

- 前言
- B站1000Wqps生产级IM服务框架
- 3590万QPS超高吞吐压测
  - 服务端配置
  - 压测参数
  - 资源使用
  - 压测结果
- goim搭建
  - 基于 docker 安装zookeeper、redis、kafka
  - 设置 env 环境变量
  - 启动 discovery
  - 启动 comet
  - 启动 logic
  - 解决exec: “gcc“: executable file not found in %PATH%问题
    - 一、什么是 MinGW-w64 ？
    - 二、为什么使用 MinGW-w64 ？
    - 三、MinGW-w64 适合做什么？
    - 四、下载和安装 MinGW-w64
    - 五、安装失败
      - 1. 问题背景
      - 2. 报错原因
      - 3. 解决方案
  - 启动 job
  - 启动 goim 的 example
- 源码解读
  - 核心依赖库
- GOIM网元架构
  - comet
  - logic
  - kafka
  - router
  - job
  - 如何进行高并发伸缩
- 几个重要的结构体
  - Bucket 结构体
  - Room 结构
  - Channel 结构
  - 消息结构
    - 1. 任务队列消息
    - 2. GOIM消息协议
- 消息流转
  - 生成消息
  - 消息的投递
- 第一棒：Logic服务的消息消峰
  - logic
  - http 服务
  - rpc 服务
  - MQ 消费处理
    - 用户消息
    - 房间消息
    - 广播消息
- 第二棒：传输消息
  - job 组件
  - job 发送消息（普通消息，房间消息，广播）
  - goroutine 和 channel 实现高并发架构
  - 队列缓存+批量写入高并发架构
  - Job 处理房间消息小结：
- 第三棒：由Comet将消息发送给客户端
  - Comet（彗星）的主要功能
  - Comet 处理的main方法
  - Comet 如何管理用户端的长连接
  - Comet 如何进行channel管理
  - 锁分段的架构
  - goroutine 和 channel 实现高并发
  - 高性能池化架构
- Redis 与 Session 结构
- 服务发现
- GOIM 的高并发架构
- GOIM 的总结
- 参考文档
- 说在最后：有问题可以找老架构尼恩取经
- 11个技术圣经 PDF

▌B站1000Wqps生产级IM服务框架：

goim是bilibili公司技术总监毛剑创作，使用go语言开发，用于B站生产线上的IM服务框架（聊天室），

官网：https://goim.io/

下面是官方的3590万QPS超高吞吐压测

▌3590万QPS超高吞吐压测：

▌服务端配置

CPU

内存

操作系统

数量

Intel(R) Xeon(R) CPU E5-2630 v2 @ 2.60GHz

DDR3 32GB

Debian GNU/Linux 8

▌压测参数

不同UID同房间在线人数: 1,000,000
持续推送时长: 15分钟
持续推送数量: 40条/秒
推送内容:
推送类型: 单房间推送
到达计算方式: 1秒统计一次,共30次

▌资源使用

每台服务端CPU使用: 2000%~2300%(刚好满负载)
每台服务端内存使用: 14GB左右
GC耗时: 504毫秒左右
流量使用: Incoming(450MBit/s), Outgoing(4.39GBit/s)

▌压测结果

推送到达: 3590万/秒左右;

其框架原理图如下：下面是官方的架构图

接下来，尼恩首先给大家介绍 goim的实操，再介绍底层架构、核心源码、高性能架构设计。

▌goim搭建：

▌基于 docker 安装zookeeper、redis、kafka

zookeeper

使用尼恩地表最强环境中现成服务

redis：

使用尼恩地表最强环境中现成服务

kafka：

version: &39;
services:
  kafka:
    image: &39;
    ports:
      - &39;
      - &39;
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
       允许使用PLAINTEXT协议(镜像中默认为关闭,需要手动开启)
      - ALLOW_PLAINTEXT_LISTENER=yes
       全局消息过期时间 6 小时(测试时可以设置短一点)
      - KAFKA_CFG_LOG_RETENTION_HOURS=6
      volumes:
       Web 管理界面 （用KnowStreaming可以不用下面的）
  kafka_manager:
    image: &39;
    ports:
      - &34;
    environment:
      ZK_HOSTS: &34;
      APPLICATION_SECRET: letmein
    depends_on:
      - kafka
34;base-env-network& vi /etc/profile
[root@centos1 discovery]39;ws://127.0.0.1:3102/sub&39;mid message&39;http://192.168.56.1:3111/goim/push/mids?operation=1000&mids=123&34;&39;mid message&39;broadcast message&39;broadcast message&34;goim-logic [version: %s env: %+v] start&34;/goim&34;/push/keys&34;/push/mids&34;/push/room&34;/push/all&34;/online/top&34;/online/room&34;/online/total&34;/nodes/weighted&34;/nodes/instances&消息格式34;PushMsg.send(push pushMsg:%v) error(%v)&34;PushMsg.send(push pushMsg:%v) error(%v)&34;goim-job [version: %s env: %+v] start&34;no match push type: %s&34;c.Push(%v) serverID:%s error(%v)&34;pushKey:%s comets:%d&34;c.Broadcast(%v) serverID:%s error(%v)&34;broadcast comets:%d", len(comets))
    return
}

房间消息处理：

getRoom(roomID) -> room.Push() -> p -> room.proto
    |
    |---> NewRoom(batch, duration)
            |
            |---> go room.pushproc() -> p <- room.proto
// goim/internal/job/room.go
type Room struct {
    c     *conf.Room // 关于房间的配置
    job   *Job       // 绑定job，为了追溯Room所属的Job
    id    string     // 房间ID
    proto chan *comet.Proto // 有缓冲channel
}

// pushproc merge proto and push msgs in batch.
// 默认batch = 20, sigTime = 1s
func (r *Room) pushproc(batch int, sigTime time.Duration) {
    var (
        n    int
        last time.Time
        p    *comet.Proto
        buf  = bytes.NewWriterSize(int(comet.MaxBodySize)) // 4096B = 4KB
    )

    // 设置了一个定时器,在一定时间后往room.proto放送一个roomReadyProto信号。
    td := time.AfterFunc(sigTime, func() {
        select {
        case r.proto <- roomReadyProto:
        default:
        }
    })
    defer td.Stop()

    for {
        if p = <-r.proto; p == nil {
            // 如果创建了room，但是读到空包
            break // exit
        } else if p != roomReadyProto {
            // 读取room.proto 如果是正常的数据包，则合并到buf中去,如果满了怎么办？
            p.WriteTo(buf)
            // 如果是第一个数据包，则重置定时器，并继续读取后续数据包
            if n++; n == 1 {
                last = time.Now()
                td.Reset(sigTime)
                continue
            } else if n < batch {
                // 后续的数据包，不会重置定时器，但是如果时间仍在第一个数据包的 sigTime 时间间隔内
                // 简单说，定时器还没到时间
                if sigTime > time.Since(last) {
                    continue
                }
            }
            // 累计的数据包数量已经超过了batch, 执行发送动作
        } else {
            // 定时器到读到了roomReadyProto 
            // 如果buf已经被重置了，则跳出循环执行清理动作；否则执行发送消息     
            if n == 0 {
                break
            }
        }

        // 发送房间内的消息
        _ = r.job.broadcastRoomRawBytes(r.id, buf.Buffer())     
        buf = bytes.NewWriterSize(buf.Size())
        n = 0

        // 如果配置了房间最大闲置时间，则重新设定定时器
        // 也就是说，如果房间被创建后，处理完了该房间的消息，并不是直接跳出循环清理房间
        // 而是，会阻塞等待下一次的消息再来，如果在 “1m / r.c.Idle” 时间内没有来，则会跳出循环清理掉该房间
        // 如果在 “1m / r.c.Idle” 内有消息，则会重新设定定时器为sigTime，并为proto计数
        if r.c.Idle != 0 {
            td.Reset(time.Duration(r.c.Idle)) // 默认15分钟
        } else {
            td.Reset(time.Minute)
        }
    }

    // 清理动作
    r.job.delRoom(r.id)
}

▌goroutine 和 channel 实现高并发架构