rabbitmq(广播方式)

通过休息队列控制用户上线下线

1. 创建交换机

@Bean(name="oninechange")
	public FanoutExchange onlineExchange() {
		log.info("【交换机实例{}创建成功】", FANOUT_EXCHANGE_NAME);
		return new FanoutExchange(FANOUT_EXCHANGE_NAME);
	}

2.创建队列

@Bean(name="onlinequeue")
	public Queue onlineQueue() {
		log.info("【队列{}实例创建成功】", MYSESSIONQUEUENAME);
		return new Queue(MYSESSIONQUEUENAME, false);
	}

.

3.绑定队列到交换机

@Bean(name="binginline")
	public Binding binglineQueue1ToExchange() {
		log.info("【绑定队列{}到交换机{}成功】", MYSESSIONQUEUENAME, FANOUT_EXCHANGE_NAME);
		return BindingBuilder.bind(onlineQueue()).to(onlineExchange());
	}

4.添加消息处理方式，监听。

@Bean(name="onlinemessageContainer")
	public SimpleMessageListenerContainer messageContainer(SessionSynchronizeReceiver SessionReceiver,
			ConnectionFactory connectionFactory) {
		// 加载处理消息A的队列
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
		// 设置接收多个队列里面的消息，这里设置接收队列A
		// 假如想一个消费者处理多个队列里面的信息可以如下设置：
		// container.setQueues(queueA(),queueB(),queueC());
		container.setQueues(onlineQueue());
		container.setExposeListenerChannel(true);
		// 设置最大的并发的消费者数量
		container.setMaxConcurrentConsumers(10);
		// 最小的并发消费者的数量
		container.setConcurrentConsumers(1);
		// 设置确认模式手工确认
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		container.setMessageListener(SessionReceiver);
		return container;
	}

5.创建监听，接受消息队列信息

@Component
	public class SessionSynchronizeReceiver implements ChannelAwareMessageListener {
		@Override
		public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
			log.debug("receiver {},channel :{}", message, channel);
			try {
				String str = new String(message.getBody(), message.getMessageProperties().getContentEncoding());
				SimpleEntry<String, Object> sim = protostuffSerializeUtil.deserializeByStr(str);
				log.debug("sim  k:{} v :{}", sim.getEntity1(), sim.getEntity2());
				if (StringUtils.equals(MYSESSIONQUEUENAME, sim.getEntity1())) {
					String token =(String) sim.getValue();
					if(!StringUtils.isBlank(token)){
					//Login login = loginMapper.selectByPrimaryKey(token);
						Object obj = jcClient.get(RedisConstant.user_online_status+"@"+token);
						if(obj==null){
							final Long time = System.currentTimeMillis();
							jcClient.put(RedisConstant.user_online_status+"@"+token,time);
							loginMapper.online(token);
							return ;
						}
					}	
					return;
				}
				//log.info("synchronize session {}", Objs.toString(sim.getEntity2()));
				//sessionEngine.local().remove(Objs.toString(sim.getEntity2()));
			} catch (Exception e) {
				log.error("", e);
			} finally {
				channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
			}
		}

	}

6。创建发送消息到指定交换机（所属队列）

/** 上线 */
	public boolean online(String access_token) {
		//QueryHelper.update(" UPDATE LOGIN SET uonline ='1' WHERE access_token =?   ", access_token);
		rabbitTemplate.convertAndSend(FANOUT_EXCHANGE_NAME, "",
				protostuffSerializeUtil.serializeToStr(new SimpleEntry<>(MYSESSIONQUEUENAME, access_token)));
		return true;
	}