首页 > 其他分享 >Android Spingboot 实现SSE通信案例

Android Spingboot 实现SSE通信案例

时间:2024-07-24 15:41:49浏览次数:13  
标签:String Spingboot SS SseEmitter E通 import sse new emitter

SSE

SSE(Server-Sent Events)是一种用于实现服务器主动向客户端推送数据的技术,它基于 HTTP 协议,利用了其长连接特性,在客户端与服务器之间建立一条持久化连接,并通过这条连接实现服务器向客户端的实时数据推送。

Server-Sent Events (SSE) 和 Sockets 都可以用于实现服务器向客户端推送消息的实时通信,差异对比:
SSE:

优点:
使用简单,只需发送 HTTP 流式响应。
自动处理网络中断和重连。
支持由浏览器原生实现的事件,如 "error" 和 "message"。

缺点:
单向通信,服务器只能发送消息给客户端。
每个连接需要服务器端的一个线程或进程。

Socket:

优点:
双向通信,客户端和服务器都可以发送或接收消息。
可以处理更复杂的应用场景,如双向对话、多人游戏等。
服务器可以更精细地管理连接,如使用长连接或短连接。

缺点:
需要处理网络中断和重连,相对复杂。
需要客户端和服务器端的代码都能处理 Socket 通信。
对开发者要求较高,需要对网络编程有深入了解。

SSE使用场景:

使用场景主要包括需要服务器主动向客户端推送数据的应用场景,‌如AI问答聊天、实时新闻、‌股票行情等。

案例

服务端基于springboot实现,默认支持SSE;
Android客户端基于OkHttp实现,同样也支SSE;

服务端接口开发

SSEController.java

package com.qxc.server.controller.sse;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


@RestController
@RequestMapping("/sse")
public class SSEController {
    Logger logger = LoggerFactory.getLogger(SSEController.class);
    public static Map<String, SseEmitter> sseEmitters = new ConcurrentHashMap<>();

    /**
     * 接收sse请求,异步处理,分批次返回结果,然后关闭SseEmitter
     * @return SseEmitter
     */
    @GetMapping("/stream-sse")
    public SseEmitter handleSse() {
        SseEmitter emitter = new SseEmitter();
        // 在新线程中发送消息,以避免阻塞主线程
        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    Map<String, Object> event = new HashMap<>();
                    String mes = "Hello, SSE " + (i+1);
                    event.put("message", mes);
                    logger.debug("emitter.send:  "+mes);
                    emitter.send(event);
                    Thread.sleep(200);
                }
                emitter.complete(); // 完成发送
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e); // 发送错误
            }
        }).start();
        return emitter;
    }

    /**
     * 接收sse请求,异步处理,分批次返回结果,并存储SseEmitter,可通过外界调用sendMsg接口,继续返回结果
     * @param uid 客户唯一标识
     * @return SseEmitter
     */
    @GetMapping("/stream-sse1")
    public SseEmitter handleSse1(@RequestParam("uid") String uid) {
        SseEmitter emitter = new SseEmitter();
        sseEmitters.put(uid, emitter);
        // 在新线程中发送消息,以避免阻塞主线程
        new Thread(() -> {
            try {
                for (int i = 10; i < 15; i++) {
                    Map<String, Object> event = new HashMap<>();
                    String mes = "Hello, SSE " + (i+1);
                    event.put("message", mes);
                    logger.debug("emitter.send:  "+mes);
                    emitter.send(event);
                    Thread.sleep(200); // 每2秒发送一次
                }
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e); // 发送错误
            }
        }).start();
        return emitter;
    }

    /**
     * 外界调用sendMsg接口,根据标识获取缓存的SseEmitter,继续返回结果
     * @param uid 客户唯一标识
     */
    @GetMapping("/sendMsg")
    public void sendMsg(@RequestParam("uid") String uid) {
        logger.debug("服务端发送消息 to " + uid);
        SseEmitter emitter = sseEmitters.get(uid);
        if(emitter != null){
            new Thread(() -> {
                try {
                    for (int i = 20; i < 30; i++) {
                        Map<String, Object> event = new HashMap<>();
                        String mes = "Hello, SSE " + (i+1);
                        event.put("message", mes);
                        logger.debug("emitter.send:  "+mes);
                        emitter.send(event);
                        Thread.sleep(200); // 每2秒发送一次
                    }
                    emitter.send(SseEmitter.event().name("stop").data(""));
                    emitter.complete(); // close connection
                    logger.debug("服务端主动关闭了连接 to " + uid);
                } catch (IOException | InterruptedException e) {
                    emitter.completeWithError(e); // error finish
                }
            }).start();
        }
    }
}

代码定义了3个接口,主要实现了两个功能:
stream-sse 接口

用于模拟一次请求,批次返回结果,然后结束SseEmitter;

stream-sse1接口 & sendMsg接口

用于模拟一次请求,批次返回结果,缓存SseEmitter,后续还可以通过sendMsg接口,通知服务端继续返回结果;

客户端功能开发

Android客户端依赖OkHttp:

implementation 'com.squareup.okhttp3:okhttp:4.9.1'
implementation("com.squareup.okhttp3:okhttp-sse:4.9.1")

布局文件:activity_main.xml

<?xml version="1.0" encoding="utf-8"?>
<RelativeLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">

    <TextView
        android:id="@+id/tv"
        android:layout_above="@id/btn"
        android:layout_centerHorizontal="true"
        android:text="--"
        android:lines="15"
        android:gravity="center"
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:layout_margin="20dp"/>

    <Button
        android:layout_width="200dp"
        android:layout_height="50dp"
        android:id="@+id/btn"
        android:text="测试普通接口"
        android:layout_centerInParent="true"/>

    <Button
        android:layout_width="200dp"
        android:layout_height="50dp"
        android:id="@+id/btn1"
        android:layout_below="@id/btn"
        android:text="sse连接"
        android:layout_centerInParent="true"/>

    <Button
        android:layout_width="200dp"
        android:layout_height="50dp"
        android:id="@+id/btn2"
        android:layout_below="@id/btn1"
        android:text="sse连接,携带参数"
        android:layout_centerInParent="true"/>
</RelativeLayout>

MainActivity.java

package com.cb.testsd;

import android.app.Activity;
import android.os.Bundle;
import android.view.View;
import android.widget.Button;
import android.widget.TextView;

import java.util.concurrent.TimeUnit;

import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.internal.sse.RealEventSource;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;

public class MainActivity extends Activity {
    Button btn;
    Button btn1;
    Button btn2;
    TextView tv;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        btn = findViewById(R.id.btn);
        btn1 = findViewById(R.id.btn1);
        btn2 = findViewById(R.id.btn2);
        tv = findViewById(R.id.tv);

        btn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        testDate();
                    }
                }).start();
            }
        });
        btn1.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        sse();
                    }
                }).start();
            }
        });
        btn2.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        sseWithParams();
                    }
                }).start();
            }
        });
    }

    private void testDate(){
        OkHttpClient client = new OkHttpClient.Builder()
                .connectTimeout(10, TimeUnit.SECONDS)   // 建立连接的超时时间
                .readTimeout(10, TimeUnit.MINUTES)  // 建立连接后读取数据的超时时间
                .build();
        Request request = new Request.Builder()
                .url("http://192.168.43.102:58888/common/getCurDate")
                .build();
        okhttp3.Call call = client.newCall(request);
        try {
            Response response = call.execute(); // 同步方法
            if (response.isSuccessful()) {
                String responseBody = response.body().string(); // 获取响应体
                System.out.println(responseBody);
                tv.setText(responseBody);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    void sse(){
        Request request = new Request.Builder()
                .url("http://192.168.43.102:58888/sse/stream-sse")
                .addHeader("Authorization", "Bearer ")
                .addHeader("Accept", "text/event-stream")
                .build();

        OkHttpClient okHttpClient = new OkHttpClient.Builder()
                .connectTimeout(10, TimeUnit.SECONDS)   // 建立连接的超时时间
                .readTimeout(10, TimeUnit.MINUTES)  // 建立连接后读取数据的超时时间
                .build();

        RealEventSource realEventSource = new RealEventSource(request, new EventSourceListener() {
            @Override
            public void onEvent(EventSource eventSource, String id, String type, String data) {
                System.out.println(data);   // 请求到的数据
                String text = tv.getText().toString();
                tv.setText(data+"\n"+text);
                if ("finish".equals(type)) {    // 消息类型,add 增量,finish 结束,error 错误,interrupted 中断

                }
            }
        });
        realEventSource.connect(okHttpClient);
    }

    void sseWithParams(){
        Request request = new Request.Builder()
                .url("http://192.168.43.102:58888/sse/stream-sse1?uid=1")
                .addHeader("Authorization", "Bearer ")
                .addHeader("Accept", "text/event-stream")
                .build();

        OkHttpClient okHttpClient = new OkHttpClient.Builder()
                .connectTimeout(10, TimeUnit.SECONDS)   // 建立连接的超时时间
                .readTimeout(10, TimeUnit.MINUTES)  // 建立连接后读取数据的超时时间
                .build();

        RealEventSource realEventSource = new RealEventSource(request, new EventSourceListener() {
            @Override
            public void onEvent(EventSource eventSource, String id, String type, String data) {
                System.out.println(data);   // 请求到的数据
                String text = tv.getText().toString();
                tv.setText(data+"\n"+text);
            }
        });
        realEventSource.connect(okHttpClient);
    }
}

效果测试

调用stream-sse接口

服务器分批次返回了结果:

调用stream-sse1接口

服务器分批次返回了结果:

通过h5调用sendMsg接口,服务端继续返回结果:

标签:String,Spingboot,SS,SseEmitter,E通,import,sse,new,emitter
From: https://www.cnblogs.com/qixingchao/p/18320799

相关文章

  • ssm古诗和诗人的可视化分析和信息检索-计算机毕业设计源码08278
    目录1绪论1.1选题背景1.2选题的目的意义1.3论文结构与章节安排2系统分析2.1.1技术可行性分析2.1.2 经济可行性分析2.1.3法律可行性分析2.2系统流程分析2.2.1添加信息流程2.2.2修改信息流程2.2.3删除信息流程2.3 系统功能分析2.3.1功能......
  • object dict cannot be used in await expression报错解释
    报错解释:这个错误通常出现在使用Python的异步编程模型时,尝试在一个不支持异步的对象上使用await关键字。在Python中,await关键字只能在异步函数中使用,而异步函数通常定义在asyncdef语句中。错误"objectdictcannotbeusedinawaitexpression"意味着你正尝试在一个普通的字典......
  • 熵,交叉熵,Focalloss
    https://zhuanlan.zhihu.com/p/149186719香农首次提出熵的定义,目的寻找一种高效/无损地编码信息的方法:以编码后数据的平均长度来衡量高效性,平均长度越小越高效;同时还需满足“无损”的条件,即编码以后不能呢个有原始信息的丢失。所以熵的定义是:无损编码事件信息的最小平均编码长度......
  • vue的数组变化侦测/计算属性/Class绑定/Style绑定
    1.数组变化侦测效果图单击push按钮后可增加其内容,直接显示在页面中单击concat按钮后修改数组,不直接显示在页面中,将数组赋值后显示addArrayt(){//不引起ul自动更新this.names.concat(["sakura"])//若不赋值则无法显示this.names=this.names.concat(["sakura"])}2.计......
  • 【计算机毕业设计】ssm499智能社区管理系统的设计与实现+vue
    现代经济快节奏发展以及不断完善升级的信息化技术,让传统数据信息的管理升级为软件存储,归纳,集中处理数据信息的管理方式。本智能社区管理系统就是在这样的大环境下诞生,其可以帮助管理者在短时间内处理完毕庞大的数据信息,使用这种软件工具可以帮助管理人员提高事务处理效率,达......
  • 【计算机毕业设计】ssm497网上茶叶销售平台设计与开发+jsp
    计算机网络发展到现在已经好几十年了,在理论上面已经有了很丰富的基础,并且在现实生活中也到处都在使用,可以说,经过几十年的发展,互联网技术已经把地域信息的隔阂给消除了,让整个世界都可以即时通话和联系,极大的方便了人们的生活。所以说,网上茶叶销售平台用计算机技术来进行设计,......
  • 【计算机毕业设计】ssm495题库管理系统的设计与实现+jsp
    随着考试的增多,需要管理的试卷越来越多。现在大多数学校考试的方式采用老师出卷安排考试时间,学生参与的方式。这种方式效率低、灵活性低,每门课程的考卷组成就给老师的统计整理工作造成困难。目前,网络发展成熟,各类管理系统层出不穷,如果可以有专业的管理系统提供网上考试,则可......
  • 【计算机毕业设计】ssm485科研经费管理系统研究与开发+vue
    身处网络时代,随着网络系统体系发展的不断成熟和完善,人们的生活也随之发生了很大的变化,人们在追求较高物质生活的同时,也在想着如何使自身的精神内涵得到提升,而读书就是人们获得精神享受非常重要的途径。为了满足人们随时随地只要有网络就可以看书的要求,科研经费管理系统被开......
  • 基于SSM的家政服务管理系统【附源码+文档】
    ......
  • 基于SSM的高校课程评价系统【附源码+文档】
    ......