开源直播即时通讯源码基于Java/Netty高性能架构支持群聊/单聊

public class ImServerBootstrap {
    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .option(ChannelOption.SO_BACKLOG, 1024)
                     .childOption(ChannelOption.TCP_NODELAY, true)
                     .childOption(ChannelOption.SO_KEEPALIVE, true)
                     .childHandler(new ImChannelInitializer());
                     
            ChannelFuture future = bootstrap.bind(8888).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class ImChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        // 空闲状态检测（读/写/读写空闲）
        pipeline.addLast(new IdleStateHandler(60, 30, 0, TimeUnit.SECONDS));
        // 自定义协议编解码
        pipeline.addLast(new ImMessageDecoder());
        pipeline.addLast(new ImMessageEncoder());
        // 心跳处理器
        pipeline.addLast(new HeartbeatHandler());
        // 鉴权处理器
        pipeline.addLast(new AuthHandler());
        // 核心业务处理器
        pipeline.addLast(new MessageHandler());
    }
}

syntax = "proto3";
package com.im.protocol;

message ImMessage {
  int64 msgId = 1;
  int32 cmd = 2; 
  int64 senderId = 3;
  int64 receiverId = 4; 
  int64 groupId = 5;   
  string content = 6;
  int64 timestamp = 7;
  int32 msgType = 8; 
}

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                System.out.println("连接读空闲，关闭通道: " + ctx.channel().id());
                ctx.close();
            }
        }
    }
}

public class ChannelManager {
    private static final ConcurrentHashMap<Long, Channel> USER_CHANNELS = new ConcurrentHashMap<>();
    
    public static void bind(long userId, Channel channel) {
        USER_CHANNELS.put(userId, channel);
        // 同步写入Redis供集群路由查询
        RedisClient.set("route:" + userId, getServerNodeId());
    }
    
    public static Channel getChannel(long userId) {
        return USER_CHANNELS.get(userId);
    }
}

public class MessageHandler extends SimpleChannelInboundHandler<ImMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ImMessage msg) {
        if (msg.getCmd() == 1) { // 单聊消息
            long receiverId = msg.getReceiverId();
            Channel receiverChannel = ChannelManager.getChannel(receiverId);
            
            if (receiverChannel != null && receiverChannel.isActive()) {
                receiverChannel.writeAndFlush(msg);
            } else {
                // 用户离线，存入离线消息队列
                OfflineMessageStore.save(msg);
            }
            
            // 构建并返回ACK
            ImMessage ack = ImMessage.newBuilder()
                    .setMsgId(msg.getMsgId())
                    .setCmd(2) // ACK指令
                    .build();
            ctx.writeAndFlush(ack);
            
            // 异步落库
            MessageService.asyncSave(msg);
        }
    }
}

public void handleGroupMessage(ImMessage msg) {
    long groupId = msg.getGroupId();
    // 1. 获取群成员列表
    List<Long> members = GroupService.getMembers(groupId);
    
    // 2. 将消息推入MQ，由多个消费者并行处理扇出
    for (Long memberId : members) {
        MqProducer.send("group_msg_topic", msg.toByteArray(), memberId.toString());
    }
}

public class GroupMessageConsumer {
    public void consume(byte[] body) {
        ImMessage msg = ImMessage.parseFrom(body);
        Channel channel = ChannelManager.getChannel(msg.getReceiverId());
        if (channel != null) {
            channel.writeAndFlush(msg);
        } else {
            OfflineMessageStore.save(msg);
        }
    }
}