1.首先我们得创建一个springboot的项目，目录结构如下

2.配置依赖

3.结构如下图

4.编写controller

package com.controller;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import com.service.DirectMessageService;

@RestController

@RequestMapping("/productMessage")

public class WebSocketController {

 equestMapping("/getMsg")

}

5.编写service

package com.service;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import com.dao.DirectMessageDao;

@Component

public class DirectMessageService {

}

6.编写dao

package com.dao;  

import java.io.IOException;

import java.util.Random;  

import java.util.UUID;

import java.util.concurrent.TimeoutException;

import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;  

import com.rabbitmq.client.Connection;  

import com.rabbitmq.client.ConnectionFactory;  

@Component

public class DirectMessageDao  

{  

 ws java.io.IOException, Exception  

 ();  

 el.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());  

 annel = connection.createChannel();  

 verity, null, message.getBytes());  

}

7.编写MyWebSocket

package com.websocket;

import java.io.IOException;

import java.util.concurrent.CopyOnWriteArraySet;

import javax.websocket.EndpointConfig;

import javax.websocket.OnClose;

import javax.websocket.OnError;

import javax.websocket.OnMessage;

import javax.websocket.OnOpen;

import javax.websocket.Session;

import javax.websocket.server.PathParam;

import javax.websocket.server.ServerEndpoint;

//该注解用来指定一个URI，客户端可以通过这个URI来连接到WebSocket。类似Servlet的注解mapping。无需在web.xml中配置。 

import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.QueueingConsumer;

@ServerEndpoint(value = "/websocket/{username}")

@Component

public class MyWebSocket {

 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。

  static int onlineCount = 0;

 包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。

 与某个客户端的连接会话，需要通过它来给客户端发送数据

  public static String username = null;

 连接建立成功调用的方法*/

        webSocketSet.add(this);     //加入set中

        addOnlineCount();           //在线数加1

 有新连接加入！当前在线人数为" + getOnlineCount());

 ctory();  

 声明direct类型转发器  

 ME, "direct");  

 指定binding_key  

 for(MyWebSocket item: MyWebSocket.webSocketSet){

 异常");

 连接关闭调用的方法

 从set中删除

        subOnlineCount();           //在线数减1

 有一连接关闭！当前在线人数为" + getOnlineCount());

 收到客户端消息后调用的方法

 客户端发送过来的消息*/

 来自客户端的消息:" + message);

 群发消息

  (IOException e) {

 发生错误时调用

 */

 发生错误");

 群发自定义消息

 {

}

8.编写GetHttpSessionConfigurator

package com.websocket;

import javax.servlet.http.HttpSession;

import javax.websocket.HandshakeResponse;

import javax.websocket.server.HandshakeRequest;

import javax.websocket.server.ServerEndpoint;

import javax.websocket.server.ServerEndpointConfig;

import javax.websocket.server.ServerEndpointConfig.Configurator;

public class GetHttpSessionConfigurator extends ServerEndpointConfig.Configurator

{

}

10.编写WebSocketConfig

package com;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration  

public class WebSocketConfig {  

}  

11.当然， Erlang和RabbitMQ服务要提前安装好。

在运行之前我们要启动该服务，

12.双机即可

13.之后就可以启动项目了。生产，并且消费了。下载本文