forked from chenxuebing3/learning
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
267 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
25 changes: 25 additions & 0 deletions
25
WebSocket/ws-push/src/main/java/com/zhangln/push/wspush/config/MqConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| package com.zhangln.push.wspush.config; | ||
|
|
||
| import org.springframework.amqp.core.Queue; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| /** | ||
| * @author sherry | ||
| * @description | ||
| * @date Create in 2019/12/9 | ||
| * @modified By: | ||
| */ | ||
| @Configuration | ||
| public class MqConfig { | ||
|
|
||
| /** | ||
| * 通用推送队列 | ||
| * @return | ||
| */ | ||
| @Bean | ||
| public Queue queue(){ | ||
| return new Queue("push-common",false,false,true); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
95 changes: 95 additions & 0 deletions
95
WebSocket/ws-push/src/main/java/com/zhangln/push/wspush/mq/PushListener.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| package com.zhangln.push.wspush.mq; | ||
|
|
||
| import com.alibaba.fastjson.JSONObject; | ||
| import com.rabbitmq.client.Channel; | ||
| import com.zhangln.push.wspush.controller.service.WsPushService; | ||
| import com.zhangln.push.wspush.vo.HttpResVo; | ||
| import com.zhangln.push.wspush.vo.HttpWsPushCondition; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.amqp.core.ExchangeTypes; | ||
| import org.springframework.amqp.core.Message; | ||
| import org.springframework.amqp.rabbit.annotation.*; | ||
| import org.springframework.amqp.rabbit.core.RabbitTemplate; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| import java.io.ByteArrayInputStream; | ||
| import java.io.ObjectInputStream; | ||
|
|
||
| /** | ||
| * @author sherry | ||
| * @description | ||
| * @date Create in 2019/12/9 | ||
| * @modified By: | ||
| */ | ||
|
|
||
| @Slf4j | ||
| @Component | ||
| public class PushListener { | ||
| @Autowired | ||
| private RabbitTemplate rabbitTemplate; | ||
| @Autowired | ||
| private WsPushService wsPushService; | ||
|
|
||
| /** | ||
| * 通用异步推送 | ||
| * 为了避免重复推送，所以这里的MQ采用工作队列模式 | ||
| * | ||
| * @param message | ||
| * @param channel | ||
| */ | ||
| @RabbitHandler | ||
| @RabbitListener(queues = "push-common") | ||
| public void commomPush(Message message, Channel channel) { | ||
| byte[] body = message.getBody(); | ||
| try (ByteArrayInputStream bais = new ByteArrayInputStream(body); | ||
| ObjectInputStream oii = new ObjectInputStream(bais);) { | ||
| Object o = oii.readObject(); | ||
| //再做一次强制类型转化，就可以拿到生产者发送的对象数据了 | ||
| HttpWsPushCondition httpWsPushCondition = (HttpWsPushCondition) o; | ||
| log.info("处理一次异步推送：{}", JSONObject.toJSONString(httpWsPushCondition)); | ||
| HttpResVo httpResVo = wsPushService.commonPushService(httpWsPushCondition); | ||
| // 广播本次推送结果 | ||
| rabbitTemplate.convertAndSend("push-result", "", httpResVo); | ||
| } catch (Exception e) { | ||
| log.error(e.getMessage(), e); | ||
| } finally { | ||
| try { | ||
| // 返回确认状态 | ||
| channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); | ||
| } catch (Exception e) { | ||
| log.error("RabbitMQ 网关日志ACK异常", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * 监听广播中的推送结果 | ||
| * | ||
| * @param message | ||
| * @param channel | ||
| */ | ||
| @RabbitListener(bindings = @QueueBinding( | ||
| value = @Queue(value = "push-result-1", durable = "false", autoDelete = "true"), | ||
| exchange = @Exchange(value = "push-result", type = ExchangeTypes.FANOUT) | ||
| )) | ||
| public void pushResult(Message message, Channel channel) { | ||
| byte[] body = message.getBody(); | ||
| try (ByteArrayInputStream bais = new ByteArrayInputStream(body); | ||
| ObjectInputStream oii = new ObjectInputStream(bais);) { | ||
| Object o = oii.readObject(); | ||
| HttpResVo httpResVo = (HttpResVo) o; | ||
| log.info("接收到一次MQ异步推送结果：", JSONObject.toJSONString(httpResVo)); | ||
| } catch (Exception e) { | ||
| log.error(e.getMessage(), e); | ||
| } finally { | ||
| try { | ||
| // 返回确认状态 | ||
| channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); | ||
| } catch (Exception e) { | ||
| log.error("RabbitMQ 网关日志ACK异常", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.