单机系统的时候，客户端和连接都有同一台服务器管理。
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