码迷,mamicode.com
首页 > 其他好文 > 详细

使用rabbitmq实现集群聊天服务器消息的路由

时间:2020-05-02 15:18:05      阅读:97      评论:0      收藏:0      [点我收藏+]

标签:local   hello   override   dom   cio   服务器   actor   管理   change   

单机系统的时候,客户端和连接都有同一台服务器管理。

 
技术图片
image.png

在本地维护一份userId到connetciont的映射

服务器可以根据userId找出对应的连接,然后把消息push出去

 

 
技术图片
image.png

但是集群环境下,连接分布在不同的机器,小明向小张发消息时跨了机器

 

 
技术图片
image.png

小明向小张发的消息,需要小张的对应连接的服务器才能推送
要完成这个需求需要解决两个问题:
1、聊天服务器这么多,怎么才能知道小张连接了哪一台机器?
2、知道是哪一台服务器,怎么才能把消息精准发送到对应的机器?

 

 
技术图片
image.png

有一种解决方案,就是把消息全量广播,然后每台机器自己去判断小张是不是在本机有连接,然后对应推送消息。但是这个方案明显不可取,因为这样会造成网络风暴,满天飞羽的消息在广播。

个人解决方案

一、对于第一个问题,怎么才能知道小张连接了哪一台机器,可以用redis去做一个map映射。
每台机器启动的时候都分配一个机器id,
当用户创建连接的时候,在redis放一个key-value,key值放的是userId,value放的是机器id。
这样就可以根据userId找对应的连接机器

 

 
技术图片
image.png

二、对于第二个问题,怎么才能把消息精准发送到对应的机器,可以让每台机器启动的时候都有一个专属的mq和专属的routingkey,使用rabbitMq的TopicExchange交换器,这样消息就能精准投递到对应的机器,routingKey可以用上面定义的机器id。
同时queue的熟悉选专属队列,这样服务器重启后,连接断开后,旧的队列会自动删除

 
技术图片
image.png

 

全过程就是:
1、小明要发消息给小张
2、根据小张userId找到小张所在的机器b
3、构建messageBody,把routingKey设置成b
4、向mq发送消息,rabbitMq根据routingKey把消息分发给机器b;

以下是代码:

1、新建springboot工程,maven如下

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

2、新建Constant做固定配置

public interface Constant {

    //队列名称,这个用uuid
    String QUEUE = "queue:"+UUID.randomUUID().toString();

    //路由键,作为机器id
    String ROUTING_KEY = "routingKey:"+new Date().getTime();
    
    //固定值,代表消息主题
    String TOPIC = "topic";
    
    //redsikey的前缀
    String REDIS_KEY = "redisKey:";
}

3、配置RabbitmqConfig

@Configuration
public class RabbitmqConfig {

    //新建topic交换器
    @Bean
    TopicExchange initExchange() {
        return new TopicExchange(Constant.TOPIC);
    }

    //新建队列,要设置独占队列,exclusive
    @Bean
    public Queue initQueue() {
        Queue queue = QueueBuilder.durable(Constant.QUEUE).exclusive().autoDelete().build();
        return queue;
    }

    //交换器和主题绑定
    @Bean
    Binding bindingiTopicExchange() {
        return BindingBuilder.bind(initQueue()).to(initExchange()).with(Constant.ROUTING_KEY);
    }

    //新建消费者
    @Bean
    public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, ChannelAwareMessageListener listener) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        // 指定消费者
        container.setMessageListener(listener);
        // 指定监听的队列
        container.setQueueNames(Constant.QUEUE);

        // 设置消费者的 ack 模式为手动确认模式
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        container.setPrefetchCount(300);
        //connection
        return container;
    }
}

4、配置消费者Consumer

@Slf4j
@Component
public class Consumer implements ChannelAwareMessageListener {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();

        // 代表投递的标识符,唯一标识了当前信道上的投递,通过 deliveryTag ,消费者就可以告诉 RabbitMQ 确认收到了当前消息,见下面的方法
        long deliveryTag = messageProperties.getDeliveryTag();

        // 如果是重复投递的消息,redelivered 为 true
        Boolean redelivered = messageProperties.getRedelivered();

        // 获取生产者发送的原始消息
        Object originalMessage = rabbitTemplate.getMessageConverter().fromMessage(message);
        log.info("接受到消息:{}",originalMessage);
        // 代表消费者确认收到当前消息,第二个参数表示一次是否 ack 多条消息
        channel.basicAck(deliveryTag, false);
    }
}

5、设置UserController

@Slf4j
@RequestMapping
@RestController
public class UserController {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMsg")
        public String sendMsg(@RequestParam Integer sendUserId,@RequestParam Integer recUserId,String msg ){
        MessageProperties messageProperties = new MessageProperties();
        String messageBody = msg;
        Message message = rabbitTemplate.getMessageConverter().toMessage(messageBody, messageProperties);
        String key = Constant.REDIS_KEY+recUserId;
        Object o = redisTemplate.opsForValue().get(key);
        if(o == null){
            throw new RuntimeException("recUserId未建立连接:"+recUserId);
        }
        rabbitTemplate.convertAndSend(Constant.TOPIC, o.toString(),message);
        return "success";
    }

    @GetMapping("/buildRoutingKey")
    public String buildRoutingKey(@RequestParam Integer recUserId){

        String key = Constant.REDIS_KEY+recUserId;
        log.info("key={},value={}",key,Constant.ROUTING_KEY);
        redisTemplate.opsForValue().set(key,Constant.ROUTING_KEY);
        return "success";
    }
}

6、实验环节,需要部署两个jar包
分别在2台服务器发起请求初始化连接。
在A服务器输入 curl localhost:8081//buildRoutingKey?recUserId=1
在B服务器输入 curl localhost:8082//buildRoutingKey?recUserId=2
然后发送消息
在任意一台服务器输入 curl localhost:8082/sendMsg?sendUserId=2&recUserId=1&msg=xiaozhangHello

在任意一台服务器输入 curl localhost:8082/sendMsg?sendUserId=1&recUserId=2&msg=xiaoMingHello

 

 
技术图片
image.png
 
技术图片
image.png

代码地址
https://github.com/hd-eujian/rabbitmq-share.git
https://gitee.com/guoeryyj/rabbitmq-share.git

使用rabbitmq实现集群聊天服务器消息的路由

标签:local   hello   override   dom   cio   服务器   actor   管理   change   

原文地址:https://www.cnblogs.com/yeyongjian/p/12817907.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!