SpringAOP+RabbitMQ+WebSocket实战详解

 更新时间:2018年11月10日 08:43:19   作者:little-sheep   我要评论

这篇文章主要介绍了SpringAOP+RabbitMQ+WebSocket实战详解,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

背景

最近公司的客户要求,分配给员工的任务除了有微信通知外,还希望PC端的网页也能实时收到通知。管理员分配任务是在我们的系统A,而员工接受任务是在系统B。两个系统都是现在已投入使用的系统。

技术选型

根据需求我们最终选用SpringAOP+RabbitMQ+WebSocket。

SpringAOP可以让我们不修改原有代码,直接将原有service作为切点,加入切面。RabbitMQ可以让A系统和B系统解耦。WebSocket则可以达到实时通知的要求。

SpringAOP

AOP称为面向切面编程,在程序开发中主要用来解决一些系统层面上的问题,比如日志,事务,权限等待。是Spring的核心模块,底层是通过动态代理来实现(动态代理将在之后的文章重点介绍)。

基本概念

Aspect(切面):通常是一个类,里面可以定义切入点和通知。

JointPoint(连接点):程序执行过程中明确的点,一般是方法的调用。

Advice(通知):AOP在特定的切入点上执行的增强处理,有before,after,afterReturning,afterThrowing,around。

Pointcut(切入点):就是带有通知的连接点,在程序中主要体现为书写切入点表达式。

通知类型

Before:在目标方法被调用之前做增强处理。

@Before只需要指定切入点表达式即可

AfterReturning:在目标方法正常完成后做增强。

@AfterReturning除了指定切入点表达式后,还可以指定一个返回值形参名returning,代表目标方法的返回值

AfterThrowing:主要用来处理程序中未处理的异常。

@AfterThrowing除了指定切入点表达式后,还可以指定一个throwing的返回值形参名,可以通过该形参名

来访问目标方法中所抛出的异常对象

After:在目标方法完成之后做增强,无论目标方法时候成功完成。

@After可以指定一个切入点表达式

Around:环绕通知,在目标方法完成前后做增强处理,环绕通知是最重要的通知类型,像事务,日志等都是环绕通知,注意编程中核心是一个ProceedingJoinPoint。

RabbitMQ

从图中我们可以看到RabbitMQ主要的结构有:Routing、Binding、Exchange、Queue。

Queue

Queue(队列)RabbitMQ的作用是存储消息,队列的特性是先进先出。

Exchange

生产者产生的消息并不是直接发送给消息队列Queue的,而是要经过Exchange(交换器),由Exchange再将消息路由到一个或多个Queue,还会将不符合路由规则的消息丢弃。

Routing

用于标记或生产者寻找Exchange。

Binding

用于Exchange和Queue做关联。

Exchange Type fanout

fanout类型的Exchange路由规则非常简单,它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

direct

direct会把消息路由到那些binding key与routing key完全匹配的Queue中。

topic

direct规则是严格意义上的匹配,换言之Routing Key必须与Binding Key相匹配的时候才将消息传送给Queue,那么topic这个规则就是模糊匹配,可以通过通配符满足一部分规则就可以传送。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息,而是根据发送的消息内容中的headers属性进行匹配。

WebSocket

了解websocket必须先知道几个常用的web通信技术及其区别。

短轮询

短轮询的基本思路就是浏览器每隔一段时间向浏览器发送http请求,服务器端在收到请求后,不论是否有数据更新,都直接进行响应。这种方式实现的即时通信,本质上还是浏览器发送请求,服务器接受请求的一个过程,通过让客户端不断的进行请求,使得客户端能够模拟实时地收到服务器端的数据的变化。

这种方式的优点是比较简单,易于理解,实现起来也没有什么技术难点。缺点是显而易见的,这种方式由于需要不断的建立http连接,严重浪费了服务器端和客户端的资源。尤其是在客户端,距离来说,如果有数量级想对比较大的人同时位于基于短轮询的应用中,那么每一个用户的客户端都会疯狂的向服务器端发送http请求,而且不会间断。人数越多,服务器端压力越大,这是很不合理的。

因此短轮询不适用于那些同时在线用户数量比较大,并且很注重性能的Web应用。

长轮询/ comet

comet指的是,当服务器收到客户端发来的请求后,不会直接进行响应,而是先将这个请求挂起,然后判断服务器端数据是否有更新。如果有更新,则进行响应,如果一直没有数据,则到达一定的时间限制(服务器端设置)后关闭连接。

长轮询和短轮询比起来,明显减少了很多不必要的http请求次数,相比之下节约了资源。长轮询的缺点在于,连接挂起也会导致资源的浪费。

SSE

SSE是HTML5新增的功能,全称为Server-Sent Events。它可以允许服务推送数据到客户端。SSE在本质上就与之前的长轮询、短轮询不同,虽然都是基于http协议的,但是轮询需要客户端先发送请求。而SSE最大的特点就是不需要客户端发送请求,可以实现只要服务器端数据有更新,就可以马上发送到客户端。

SSE的优势很明显,它不需要建立或保持大量的客户端发往服务器端的请求,节约了很多资源,提升应用性能。并且SSE的实现非常简单,不需要依赖其他插件。

WebSocket

WebSocket是Html5定义的一个新协议,与传统的http协议不同,该协议可以实现服务器与客户端之间全双工通信。简单来说,首先需要在客户端和服务器端建立起一个连接,这部分需要http。连接一旦建立,客户端和服务器端就处于平等的地位,可以相互发送数据,不存在请求和响应的区别。

WebSocket的优点是实现了双向通信,缺点是服务器端的逻辑非常复杂。现在针对不同的后台语言有不同的插件可以使用。

四种Web即时通信技术比较

从兼容性角度考虑,短轮询>长轮询>长连接SSE>WebSocket;

从性能方面考虑,WebSocket>长连接SSE>长轮询>短轮询。

实战

项目使用SpringBoot搭建。RabbitMQ的安装这里不讲述。

RabbitMQ配置

两个系统A、B都需要操作RabbitMQ,其中A生产消息,B消费消息。故都需要配置。

1、首先引入RabbitMQ的dependency:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

这个dependency中包含了RabbitMQ相关dependency。

2、在项目的配置文件里配置为使用rabbitmq及其参数。

application-pro.yml

#消息队列
message.queue.type: rabbitmq
## rabbit mq properties
rabbitmq:
 host: localhost
 port: 5672
 username: guest
 password: guest

application.properties

#将要使用的队列名
rabbitmq.websocket.msg.queue=websocket_msg_queue

3、创建配置文件。队列的创建交给spring。

RabbitMQConfig.java

@Configuration
@EnableRabbit
public class RabbitMQConfig {

  @Value("${rabbitmq.host}")
  private String host;
  @Value("${rabbitmq.port}")
  private String port;
  @Value("${rabbitmq.username}")
  private String username;
  @Value("${rabbitmq.password}")
  private String password;
  @Value("${rabbitmq.websocket.msg.queue}")
  private String webSocketMsgQueue;

  @Bean
  public ConnectionFactory connectionFactory() throws IOException {
    CachingConnectionFactory factory = new CachingConnectionFactory();
    factory.setUsername(username);
    factory.setPassword(password);
//    factory.setVirtualHost("test");
    factory.setHost(host);
    factory.setPort(Integer.valueOf(port));
    factory.setPublisherConfirms(true);

    //设置队列参数,是否持久化、队列TTL、队列消息TTL等
    factory.createConnection().createChannel(false).queueDeclare(webSocketMsgQueue, true, false, false, null);
    return factory;
  }

  @Bean
  public MessageConverter messageConverter() {
    return new Jackson2JsonMessageConverter();
  }

  @Bean
  @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  // 必须是prototype类型
  public RabbitTemplate rabbitTemplate() throws IOException {
    return new RabbitTemplate(connectionFactory());
  }

  @Bean
  public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() throws IOException {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setConcurrentConsumers(3);
    factory.setMaxConcurrentConsumers(10);
    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    return factory;
  }
}

4、系统B中创建队列监听,当队列有消息时,发送websocket通知。

RabbitMQListener.java

@Component
public class RabbitMQListener {

  @Autowired
  private RabbitMQService mqService;

  /**
   * WebSocket推送监听器
   * @param socketEntity
   * @param deliveryTag
   * @param channel
   */
  @RabbitListener(queues = "websocket_msg_queue")
  public void webSocketMsgListener(@Payload WebSocketMsgEntity socketMsgEntity, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException {
    mqService.handleWebSocketMsg(socketMsgEntity, deliveryTag, channel);
  }

}

RabbitMQService.java

public class RabbitMQService {
  @Autowired
  private MessageWebSocketHandler messageWebSocketHandler;

  /**
   * @param socketMsgEntity
   * @param deliveryTag
   * @param channel
   * @throws IOException
   */
  void handleWebSocketMsg(WebSocketMsgEntity socketMsgEntity, long deliveryTag, Channel channel) throws IOException {
    try {
      messageWebSocketHandler.sendMessageToUsers(socketMsgEntity.toJsonString(), socketMsgEntity.getToUserIds());
      channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
      channel.basicNack(deliveryTag, false, false);
    }
  }
}

WebSocketMsgEntity为MQ中传送的实体。

public class WebSocketMsgEntity implements Serializable {
  public enum OrderType{
    repair("维修"),
    maintain("保养"),
    measure("计量");

    OrderType(String value){
      this.value = value;
    }
    String value;

    public String getValue() {
      return value;
    }
  }
  //设备名称
  private String EquName;
  //设备编号
  private String EquId;
  //工单类型
  private OrderType orderType;
  //工单单号
  private String orderId;
  //工单状态
  private String orderStatus;
  //创建时间
  private Date createTime;
  //消息接收人ID
  private List<String> toUserIds;

  public String getEquName() {
    return EquName;
  }

  public void setEquName(String equName) {
    EquName = equName;
  }

  public String getOrderId() {
    return orderId;
  }

  public void setOrderId(String orderId) {
    this.orderId = orderId;
  }

  public String getEquId() {
    return EquId;
  }

  public void setEquId(String equId) {
    EquId = equId;
  }

  public String getOrderStatus() {
    return orderStatus;
  }

  public void setOrderStatus(String orderStatus) {
    this.orderStatus = orderStatus;
  }


  public OrderType getOrderType() {
    return orderType;
  }

  public void setOrderType(OrderType orderType) {
    this.orderType = orderType;
  }

  public Date getCreateTime() {
    return createTime;
  }

  public void setCreateTime(Date createTime) {
    this.createTime = createTime;
  }

  public List<String> getToUserIds() {
    return toUserIds;
  }

  public void setToUserIds(List<String> toUserIds) {
    this.toUserIds = toUserIds;
  }

  public String toJsonString(){
    return JSON.toJSONString(this);
  }
}

SpringAOP

1、系统A中创建一个切面类DataInterceptor.java

@Aspect
@Component
public class DataInterceptor {
  @Autowired
  private MessageQueueService queueService;


  //维修工单切点
  @Pointcut("execution(* com.zhishang.hes.common.service.impl.RepairServiceImpl.executeFlow(..))")
  private void repairMsg() {
  }

  /**
   * 返回通知,方法执行正常返回时触发
   *
   * @param joinPoint
   * @param result
   */
  @AfterReturning(value = "repairMsg()", returning = "result")
  public void afterReturning(JoinPoint joinPoint, Object result) {
    //此处可以获得切点方法名
    //String methodName = joinPoint.getSignature().getName();
    EquipmentRepair equipmentRepair = (EquipmentRepair) result;
    WebSocketMsgEntity webSocketMsgEntity = this.generateRepairMsgEntity(equipmentRepair);
    if (webSocketMsgEntity == null) {
      return;
    }
    queueService.send(webSocketMsgEntity);
  }

  /**
   * 生成发送到MQ的维修消息
   *
   * @param equipmentRepair
   * @return
   */
  private WebSocketMsgEntity generateRepairMsgEntity(EquipmentRepair equipmentRepair) {
    WebSocketMsgEntity webSocketMsgEntity = generateRepairMsgFromTasks(equipmentRepair);
    return webSocketMsgEntity;
  }

  /**
   * 从任务中生成消息
   *
   * @param equipmentRepair
   * @return
   */
  private WebSocketMsgEntity generateRepairMsgFromTasks(EquipmentRepair equipmentRepair) {
    //业务代码略
  }

}

2、发送消息到MQ。这里只贴了发送的核心代码

public class RabbitMessageQueue extends AbstractMessageQueue {

  @Value("${rabbitmq.websocket.msg.queue}")
  private String webSocketMsgQueue;

  @Autowired
  private RabbitTemplate rabbitTemplate;

  @Override
  public void send(WebSocketMsgEntity entity) {
    //没有指定exchange,则使用默认名为“”的exchange,binding名与queue名相同
    rabbitTemplate.convertAndSend(webSocketMsgQueue, entity);
  }
}

WebSocket

1、 系统B中引入websocket服务端dependency

<dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-websocket</artifactId>
  <version>4.3.10.RELEASE</version>
</dependency>

2、 配置websocket,添加处理类

WebSocketConfigurer.java

@Configuration
@EnableWebSocket
public class WebSocketConfig extends WebMvcConfigurerAdapter implements WebSocketConfigurer {

  private static Logger logger = LoggerFactory.getLogger(WebSocketConfig.class);

  @Override
  public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
    //配置webSocket路径
    registry.addHandler(messageWebSocketHandler(),"/msg-websocket").addInterceptors(new MyHandshakeInterceptor()).setAllowedOrigins("*");
    //配置webSocket路径 支持前端使用socketJs
    registry.addHandler(messageWebSocketHandler(), "/sockjs/msg-websocket").setAllowedOrigins("*").addInterceptors(new MyHandshakeInterceptor()).withSockJS();
  }

  @Bean
  public MessageWebSocketHandler messageWebSocketHandler() {
    logger.info("......创建MessageWebSocketHandler......");
    return new MessageWebSocketHandler();
  }

}

MessageWebSocketHandler.java 主要用于websocket连接及消息发送处理。配置中还使用了连接握手时的处理,主要是取用户登陆信息,这里不多讲述。

public class MessageWebSocketHandler extends TextWebSocketHandler {
  private static Logger logger = LoggerFactory.getLogger(SystemWebSocketHandler.class);
  private static ConcurrentHashMap<String, CopyOnWriteArraySet<WebSocketSession>> users = new ConcurrentHashMap<>();

  @Override
  public void afterConnectionEstablished(WebSocketSession session) throws Exception {
    String userId = session.getAttributes().get("WEBSOCKET_USERID").toString();
    logger.info("......AfterConnectionEstablished......");
    logger.info("session.getId:" + session.getId());
    logger.info("session.getLocalAddress:" + session.getLocalAddress().toString());
    logger.info("userId:" + userId);
    //websocket连接后记录连接信息
    if (users.keySet().contains(userId)) {
      CopyOnWriteArraySet<WebSocketSession> webSocketSessions = users.get(userId);
      webSocketSessions.add(session);
    } else {
      CopyOnWriteArraySet<WebSocketSession> webSocketSessions = new CopyOnWriteArraySet<>();
      webSocketSessions.add(session);
      users.put(userId, webSocketSessions);
    }
  }

  @Override
  public void handleTransportError(WebSocketSession session, Throwable throwable) throws Exception {
    removeUserSession(session);
    if (session.isOpen()) {
      session.close();
    }
    logger.info("异常出现handleTransportError" + throwable.getMessage());
  }

  @Override
  public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
    removeUserSession(session);
    logger.info("关闭afterConnectionClosed" + closeStatus.getReason());
  }

  @Override
  public boolean supportsPartialMessages() {
    return false;
  }

  /**
   * 给符合要求的在线用户发送消息
   *
   * @param message
   */
  public void sendMessageToUsers(String message, List<String> userIds) throws IOException{
    if (StringUtils.isEmpty(message) || CollectionUtils.isEmpty(userIds)) {
      return;
    }
    if (users.isEmpty()) {
      return;
    }
    for (String userId : userIds) {
      if (!users.keySet().contains(userId)) {
        continue;
      }
      CopyOnWriteArraySet<WebSocketSession> webSocketSessions = users.get(userId);
      if (webSocketSessions == null) {
        continue;
      }
      for (WebSocketSession webSocketSession : webSocketSessions) {
        if (webSocketSession.isOpen()) {
          try {
            webSocketSession.sendMessage(new TextMessage(message));
          } catch (IOException e) {
            logger.error(" WebSocket server send message ERROR " + e.getMessage());
            try {
              throw e;
            } catch (IOException e1) {
              e1.printStackTrace();
            }
          }
        }
      }
    }
  }

  /**
   * websocket清除连接信息
   *
   * @param session
   */
  private void removeUserSession(WebSocketSession session) {
    String userId = session.getAttributes().get("WEBSOCKET_USERID").toString();
    if (users.keySet().contains(userId)) {
      CopyOnWriteArraySet<WebSocketSession> webSocketSessions = users.get(userId);
      webSocketSessions.remove(session);
      if (webSocketSessions.isEmpty()) {
        users.remove(userId);
      }
    }
  }
}

整个功能完成后,A系统分配任务时,系统B登陆用户收到的消息如图:

总体流程:

1、对于系统B,每个登陆的用户都会和服务器建立websocket长连接。

2、系统A生成任务,AOP做出响应,将封装的消息发送给MQ。

3、系统B中的MQ监听发现队列有消息到达,消费消息。

4、系统B通过websocket长连接将消息发给指定的登陆用户。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • java验证用户是否已经登录 java实现自动登录

    java验证用户是否已经登录 java实现自动登录

    这篇文章主要介绍了java验证用户是否已经登录,java实现自动登录,感兴趣的小伙伴们可以参考一下
    2016-04-04
  • 定义hashcode时使用31系数的原因

    定义hashcode时使用31系数的原因

    这篇文章主要介绍了定义hashcode时使用31系数的原因,具有一定借鉴价值,需要的朋友可以参考下
    2018-01-01
  • Java正则多字符串匹配替换

    Java正则多字符串匹配替换

    正则表达式异常强大,一直理解不深,用的也不深,这次项目中尝试,体会到了它的强大之处。字符串查找,匹配,替换,正则无不能做,特别是灵活的运用子串匹配得到的变量值$1,$2,再进行二次处理能够达到很巧妙的效果。
    2013-02-02
  • 浅析Java中JSONObject和JSONArray使用

    浅析Java中JSONObject和JSONArray使用

    这篇文章主要介绍了Java中JSONObject和JSONArray使用的相关资料,需要的朋友可以参考下
    2016-06-06
  • Spring Boot(二)之web综合开发

    Spring Boot(二)之web综合开发

    本篇文章为大家介绍spring boot的其它特性(有些未必是spring boot体系桟的功能,但是是spring特别推荐的一些开源技术本文也会介绍),对了这里只是一个大概的介绍,特别详细的使用我们会在其它的文章中来展开说明
    2017-05-05
  • 微信跳一跳刷分java代码实现

    微信跳一跳刷分java代码实现

    这篇文章主要介绍了微信跳一跳刷分java代码实现,还为大家分享了java刷微信跳一跳问题集,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-01-01
  • 详解如何让Spring MVC显示自定义的404 Not Found页面

    详解如何让Spring MVC显示自定义的404 Not Found页面

    这篇文章主要介绍了详解如何让Spring MVC显示自定义的404 Not Found页面,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2018-10-10
  • Java的Struts框架中的if/else标签使用详解

    Java的Struts框架中的if/else标签使用详解

    这篇文章主要介绍了Java的Struts框架中的if/else标签使用详解,Struts是Java的SSH三大web开发框架之一,需要的朋友可以参考下
    2015-12-12
  • MyEclipse+Tomcat配置详解(图文)

    MyEclipse+Tomcat配置详解(图文)

    这篇文章主要介绍了MyEclipse+Tomcat配置详解(图文),小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2019-01-01
  • Java中SimpleDateFormat的使用方法

    Java中SimpleDateFormat的使用方法

    这篇文章主要为大家详细介绍了Java中SimpleDateFormat的使用方法,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2017-12-12

最新评论