服务器之家：专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - SpringBoot webSocket实现发送广播、点对点消息和Android接收

SpringBoot webSocket实现发送广播、点对点消息和Android接收

2020-08-29 14:40宁惊蛰 Java教程

这篇文章主要介绍了SpringBoot webSocket实现发送广播、点对点消息和Android接收,具有一定的参考价值，感兴趣的小伙伴们可以参考一下。

1、SpringBoot webSocket

SpringBoot 使用的websocket 协议，不是标准的websocket协议，使用的是名称叫做STOMP的协议。

1.1 STOMP协议说明

STOMP，Streaming Text Orientated Message Protocol，是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。

它提供了一个可互操作的连接格式，允许STOMP客户端与任意STOMP消息代理(Broker)进行交互，类似于OpenWire(一种二进制协议)。

由于其设计简单，很容易开发客户端，因此在多种语言和多种平台上得到广泛应用。其中最流行的STOMP消息代理是Apache ActiveMQ。

1.2 搭建

本人使用的是Inject idea 搭建的springBoot websocket，并未采用熟悉的gradle，而是采用了maven方式搭建。

项目结构如下

SpringBoot webSocket实现发送广播、点对点消息和Android接收

pom.xml：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 
 <groupId>com.drawthink</groupId>
 <artifactId>websocketdemo</artifactId>
 <version>0.0.1-SNAPSHOT</version>
 <packaging>jar</packaging>
 
 <name>webSocketdemo</name>
 <description>webSocketDemo project for Spring Boot</description>
 
 <parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>1.3.6.RELEASE</version>
  <relativePath/> <!-- lookup parent from repository -->
 </parent>
 
 <properties>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  <java.version>1.8</java.version>
 </properties>
 
 <dependencies>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-thymeleaf</artifactId>
  </dependency>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-websocket</artifactId>
  </dependency>
 
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
  </dependency>
 </dependencies>
 
 <build>
  <plugins>
   <plugin>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-maven-plugin</artifactId>
   </plugin>
  </plugins>
 </build>
 
 
</project>

Application：

?
1
2
3
4
5
6
7
8
9
10
11
12
package com.drawthink;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class WebSocketdemoApplication {
 
 public static void main(String[] args) {
  SpringApplication.run(WebSocketdemoApplication.class, args);
 }
}

WebSocketConfig

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
package com.drawthink.websocket;
 
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
 
/**
 * Created by lincoln on 16-10-25
 */
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {
 @Override
 public void registerStompEndpoints(StompEndpointRegistry stompEndpointRegistry) {
  //允许使用socketJs方式访问，访问点为hello，允许跨域
  stompEndpointRegistry.addEndpoint("/hello").setAllowedOrigins("*").withSockJS();
 }
 
 @Override
 public void configureMessageBroker(MessageBrokerRegistry registry) {
  //订阅Broker名称
  registry.enableSimpleBroker("/topic","/user");
  //全局使用的订阅前缀（客户端订阅路径上会体现出来）
  registry.setApplicationDestinationPrefixes("/app/");
  //点对点使用的订阅前缀（客户端订阅路径上会体现出来），不设置的话，默认也是/user/
  //registry.setUserDestinationPrefix("/user/");
 }
}

WebSocketController

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package com.drawthink.websocket.controller;
 
import com.drawthink.message.ClientMessage;
import com.drawthink.message.ServerMessage;
import com.drawthink.message.ToUserMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Controller;
 
/**
 * Created by lincoln on 16-10-25
 */
@Controller
public class WebSocketController {
 
 @MessageMapping("/welcome")
 //SendTo 发送至 Broker 下的指定订阅路径
 @SendTo("/topic/getResponse")
 public ServerMessage say(ClientMessage clientMessage){
  //方法用于广播测试
  System.out.println("clientMessage.getName() = " + clientMessage.getName());
  return new ServerMessage("Welcome , "+clientMessage.getName()+" !");
 }
 
 //注入SimpMessagingTemplate 用于点对点消息发送
 @Autowired
 private SimpMessagingTemplate messagingTemplate;
 
 @MessageMapping("/cheat")
 // 发送的订阅路径为/user/{userId}/message
 // /user/路径是默认的一个，如果想要改变，必须在config 中setUserDestinationPrefix
 public void cheatTo(ToUserMessage toUserMessage){
  //方法用于点对点测试
  System.out.println("toUserMessage.getMessage() = " + toUserMessage.getMessage());
  System.out.println("toUserMessage.getUserId() = " + toUserMessage.getUserId());          messagingTemplate.convertAndSendToUser(toUserMessage.getUserId(),"/message",toUserMessage.getMessage());
 }
}

Vo

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
package com.drawthink.message;
 
/**
 * Created by lincoln on 16-10-25
 */
public class ClientMessage {
 private String name;
 
 public String getName() {
  return name;
 }
 
 public void setName(String name) {
  this.name = name;
 }
}
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
package com.drawthink.message;
 
/**
 * Created by lincoln on 16-10-25
 */
public class ServerMessage {
 private String responseMessage;
 
 public ServerMessage(String responseMessage) {
  this.responseMessage = responseMessage;
 }
 
 public String getResponseMessage() {
  return responseMessage;
 }
 
 public void setResponseMessage(String responseMessage) {
  this.responseMessage = responseMessage;
 }
}
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package com.drawthink.message;
 
/**
 * Created by lincoln on 16-10-25
 */
public class ToUserMessage {
 private String userId;
 private String message;
 
 public String getUserId() {
  return userId;
 }
 
 public void setUserId(String userId) {
  this.userId = userId;
 }
 
 public String getMessage() {
  return message;
 }
 
 public void setMessage(String message) {
  this.message = message;
 }
}

Android 客户端

STOMP协议在Android系统中没有默认实现，必须自行去实现。不过好消息是，开源大神们已经完成了Android上使用STOMP协议的实现，所以我们只需要使用就好了。

地址：StompProtocolAndroid.rar

搭建

build.gradle(app)

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
apply plugin: 'com.android.application'
 
android {
 compileSdkVersion 24
 buildToolsVersion "24.0.3"
 defaultConfig {
  applicationId "com.drawthink.websocket"
  minSdkVersion 16
  targetSdkVersion 24
  versionCode 1
  versionName "1.0"
  testInstrumentationRunner "android.support.test.runner.AndroidJUnitRunner"
 }
 buildTypes {
  release {
   minifyEnabled false
   proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
  }
 }
}
 
dependencies {
 compile fileTree(include: ['*.jar'], dir: 'libs')
 androidTestCompile('com.android.support.test.espresso:espresso-core:2.2.2', {
  exclude group: 'com.android.support', module: 'support-annotations'
 })
 compile 'com.android.support:appcompat-v7:24.2.1'
 testCompile 'junit:junit:4.12'
 //依赖STOMP协议的Android实现
 compile 'com.github.NaikSoftware:StompProtocolAndroid:1.1.1'
 //StompProtocolAndroid 依赖于webSocket的标准实现
 compile 'org.java-websocket:Java-WebSocket:1.3.0'
}

接收广播实例：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
package com.drawthink.websocket;
 
import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.EditText;
import android.widget.TextView;
import android.widget.Toast;
 
import org.java_websocket.WebSocket;
 
import rx.Subscriber;
import rx.functions.Action1;
import ua.naiksoftware.stomp.LifecycleEvent;
import ua.naiksoftware.stomp.Stomp;
import ua.naiksoftware.stomp.client.StompClient;
import ua.naiksoftware.stomp.client.StompMessage;
 
import static android.content.ContentValues.TAG;
 
public class MainActivity extends AppCompatActivity {
 
 private TextView serverMessage;
 private Button start;
 private Button stop;
 private Button send;
 private EditText editText;
 private StompClient mStompClient;
 private Button cheat;
 
 @Override
 protected void onCreate(Bundle savedInstanceState) {
  super.onCreate(savedInstanceState);
  setContentView(R.layout.activity_main);
  bindView();
  start.setOnClickListener(new View.OnClickListener() {
   @Override
   public void onClick(View v) {
   //创建client 实例
    createStompClient();
   //订阅消息
    registerStompTopic();
   }
  });
 
  send.setOnClickListener(new View.OnClickListener() {
   @Override
   public void onClick(View v) {
    mStompClient.send("/app/welcome","{\"name\":\""+editText.getText()+"\"}")
      .subscribe(new Subscriber<Void>() {
     @Override
     public void onCompleted() {
      toast("发送成功");
     }
 
     @Override
     public void onError(Throwable e) {
      e.printStackTrace();
      toast("发送错误");
     }
 
     @Override
     public void onNext(Void aVoid) {
 
     }
    });
   }
  });
 
  stop.setOnClickListener(new View.OnClickListener() {
   @Override
   public void onClick(View v) {
    mStompClient.disconnect();
   }
  });
 
  cheat.setOnClickListener(new View.OnClickListener() {
   @Override
   public void onClick(View v) {
    startActivity(new Intent(MainActivity.this,CheatActivity.class));
    if(mStompClient != null) {
     mStompClient.disconnect();
    }
    finish();
   }
  });
 }
 
 private void showMessage(final StompMessage stompMessage) {
  runOnUiThread(new Runnable() {
   @Override
   public void run() {
    serverMessage.setText("stomp command is --->"+stompMessage.getStompCommand() +" body is --->"+stompMessage.getPayload());
   }
  });
 }
 
 //创建client 实例
 private void createStompClient() {
  mStompClient = Stomp.over(WebSocket.class, "ws://192.168.0.46:8080/hello/websocket");
  mStompClient.connect();
  Toast.makeText(MainActivity.this,"开始连接 192.168.0.46:8080",Toast.LENGTH_SHORT).show();
  mStompClient.lifecycle().subscribe(new Action1<LifecycleEvent>() {
   @Override
   public void call(LifecycleEvent lifecycleEvent) {
    switch (lifecycleEvent.getType()) {
     case OPENED:
      Log.d(TAG, "Stomp connection opened");
      toast("连接已开启");
      break;
 
     case ERROR:
      Log.e(TAG, "Stomp Error", lifecycleEvent.getException());
      toast("连接出错");
      break;
     case CLOSED:
      Log.d(TAG, "Stomp connection closed");
      toast("连接关闭");
      break;
    }
   }
  });
 }
 
 //订阅消息
 private void registerStompTopic() {
  mStompClient.topic("/topic/getResponse").subscribe(new Action1<StompMessage>() {
   @Override
   public void call(StompMessage stompMessage) {
    Log.e(TAG, "call: " +stompMessage.getPayload() );
    showMessage(stompMessage);
   }
  });
 
 }
 
 private void toast(final String message) {
  runOnUiThread(new Runnable() {
   @Override
   public void run() {
    Toast.makeText(MainActivity.this,message,Toast.LENGTH_SHORT).show();
   }
  });
 }
 
 private void bindView() {
  serverMessage = (TextView) findViewById(R.id.serverMessage);
  start = (Button) findViewById(R.id.start);
  stop = (Button) findViewById(R.id.stop);
  send = (Button) findViewById(R.id.send);
  editText = (EditText) findViewById(R.id.clientMessage);
  cheat = (Button) findViewById(R.id.cheat);
 }
}

点对点

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
package com.drawthink.websocket;
 
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.EditText;
import android.widget.LinearLayout;
import android.widget.TextView;
import android.widget.Toast;
 
import org.java_websocket.WebSocket;
 
import rx.Subscriber;
import rx.functions.Action1;
import ua.naiksoftware.stomp.LifecycleEvent;
import ua.naiksoftware.stomp.Stomp;
import ua.naiksoftware.stomp.client.StompClient;
import ua.naiksoftware.stomp.client.StompMessage;
 
import static android.content.ContentValues.TAG;
 
public class CheatActivity extends AppCompatActivity {
 
 private EditText cheat;
 private Button send;
 private LinearLayout message;
 private StompClient mStompClient;
 
 @Override
 protected void onCreate(Bundle savedInstanceState) {
  super.onCreate(savedInstanceState);
  setContentView(R.layout.activity_cheat);
  bindView();
  createStompClient();
  registerStompTopic();
  send.setOnClickListener(new View.OnClickListener() {
   @Override
   public void onClick(View v) {
   // 向/app/cheat发送Json数据
    mStompClient.send("/app/cheat","{\"userId\":\"lincoln\",\"message\":\""+cheat.getText()+"\"}")
      .subscribe(new Subscriber<Void>() {
       @Override
       public void onCompleted() {
        toast("发送成功");
       }
 
       @Override
       public void onError(Throwable e) {
        e.printStackTrace();
        toast("发送错误");
       }
 
       @Override
       public void onNext(Void aVoid) {
 
       }
      });
   }
  });
 }
 
 private void bindView() {
  cheat = (EditText) findViewById(R.id.cheat);
  send = (Button) findViewById(R.id.send);
  message = (LinearLayout) findViewById(R.id.message);
 }
 
 private void createStompClient() {
  mStompClient = Stomp.over(WebSocket.class, "ws://192.168.0.46:8080/hello/websocket");
  mStompClient.connect();
  Toast.makeText(CheatActivity.this,"开始连接 192.168.0.46:8080",Toast.LENGTH_SHORT).show();
  mStompClient.lifecycle().subscribe(new Action1<LifecycleEvent>() {
   @Override
   public void call(LifecycleEvent lifecycleEvent) {
    switch (lifecycleEvent.getType()) {
     case OPENED:
      Log.d(TAG, "Stomp connection opened");
      toast("连接已开启");
      break;
 
     case ERROR:
      Log.e(TAG, "Stomp Error", lifecycleEvent.getException());
      toast("连接出错");
      break;
     case CLOSED:
      Log.d(TAG, "Stomp connection closed");
      toast("连接关闭");
      break;
    }
   }
  });
 }
 
 // 接收/user/xiaoli/message路径发布的消息
 private void registerStompTopic() {
  mStompClient.topic("/user/xiaoli/message").subscribe(new Action1<StompMessage>() {
   @Override
   public void call(StompMessage stompMessage) {
    Log.e(TAG, "call: " +stompMessage.getPayload() );
    showMessage(stompMessage);
   }
  });
 }
 
 private void showMessage(final StompMessage stompMessage) {
  runOnUiThread(new Runnable() {
   @Override
   public void run() {
    TextView text = new TextView(CheatActivity.this);
    text.setLayoutParams(new LinearLayout.LayoutParams(LinearLayout.LayoutParams.MATCH_PARENT, LinearLayout.LayoutParams.WRAP_CONTENT));
    text.setText(System.currentTimeMillis() +" body is --->"+stompMessage.getPayload());
    message.addView(text);
   }
  });
 }
 
 
 private void toast(final String message) {
  runOnUiThread(new Runnable() {
   @Override
   public void run() {
    Toast.makeText(CheatActivity.this,message,Toast.LENGTH_SHORT).show();
   }
  });
 }
}

代码比较乱，说明一下。

1、STOMP 使用的时候，关键是发布订阅的关系，使用过消息队列，例如rabbitMQ的应该很容易理解。

服务器端 WebSocketConfig.Java文件控制的就是订阅发布的路径关系。

2、websocket的路径说明，本例中连接的是ws://192.168.0.46:8080/hello/websocket路径，/hello是在WebSocketConfig的stompEndpointRegistry.addEndpoint(“/hello”).setAllowedOrigins(““).withSockJS();*确定的， 如果有多个endpoint，这个地方的路径也会随之变化。

3、发布路径

发布信息的路径是由WebSocketConfig中的 setApplicationDestinationPrefixes(“/app/”); 和 Controller 中@MessageMapping(“/welcome”) 组合确定的。

例如发广播消息，路径为/app/welcome

例如发点对点消息，路径为/app/cheat

4、消息订阅路径

订阅broker源自WebSocketConfig中的registry.enableSimpleBroker(“/topic”,”/user”);此处开放了两个broker，具体的订阅服务路径给基于Controller中的 @SendTo(“/topic/getResponse”)或SimpMessagingTemplate中给定。（注：此处，服务器和客户端须约定订阅路径）

5、关于心跳

订阅发布模型的心跳很简单，客户端向一个指定的心跳路径发送心跳，服务器处理，服务器使用指定的订阅路径向客户端发心跳，即可。因为没有Socket，只需要记录是否联通的状态即可，重连客户端做一下就好了。

本人菜鸟，肯定有些地方没有搞清楚，如果有误，请大神斧正。

代码下载地址：blogRepository.rar

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

原文链接：http://blog.csdn.net/soslinken/article/details/53021510

  • Android
  • springboot
  • WebSocket
  • 发送广播
  • 点对点消息

    • 延伸 · 阅读

    精彩推荐
    最近更新
    编辑推荐
    阅读排行
    热门标签
    501 　 406 　 415 　 1433 　 2000 　 401 　 330 　 GOF23 　 new 　 newInstance 　 Hibernate 　 volatile 　 集合 　 spring 　 consul 　 critical 　 嵌套循环 　 注册服务 　 堆栈类 　 supplyAsync 　 遍历 　 设计模式 　 观察者模式 　 小数 　 JAVA8 　 写入文件 　 回车 　 制表符 　 排列组合 　 Java 　
    © 2019-2023 服务器之家(www.zzvips.com) 版权所有关于我们联系我们版权申明网站地图