Springboot 结合 Netty 实战聊天系统

发表于

音视频技术为什么需要微服务

微服务，英文名：microservice，百度百科上将其定义为：SOA 架构的一种变体。微服务（或微服务架构）是一种将应用程序构造为一组低耦合的服务。

微服务有着一些鲜明的特点：

  • 功能单一
  • 服务粒度小
  • 服务间独立性强
  • 服务间依赖性弱
  • 服务独立维护
  • 服务独立部署

对于每一个微服务来说，其提供的功能应该是单一的；其粒度很小的；它只会提供某一业务功能涉及到的相关接口。如：电商系统中的订单系统、支付系统、产品系统等，每一个系统服务都只是做该系统独立的功能，不会涉及到不属于它的功能逻辑。

微服务之间的依赖性应该是尽量弱的，这样带来的好处是：不会因为单一系统服务的宕机，而导致其它系统无法正常运行，从而影响用户的体验。同样以电商系统为例：用户将商品加入购物车后，提交订单，这时候去支付，发现无法支付，此时，可以将订单进入待支付状态，从而防止订单的丢失和用户体验的不友好。如果订单系统与支付系统的强依赖性，会导致订单系统一直在等待支付系统的回应，这样会导致用户的界面始终处于加载状态，从而导致用户无法进行任何操作。

当出现某个微服务的功能需要升级，或某个功能需要修复 bug 时，只需要把当前的服务进行编译、部署即可，不需要一个个打包整个产品业务功能的巨多服务，独立维护、独立部署。

上面描述的微服务，其实突出其鲜明特性：高内聚、低耦合，问题来了。什么是高内聚，什么是低耦合呢？所谓高内聚：就是说每个服务处于同一个网络或网域下，而且相对于外部，整个的是一个封闭的、安全的盒子。盒子对外的接口是不变的，盒子内部各模块之间的接口也是不变的，但是各模块内部的内容可以更改。模块只对外暴露最小限度的接口，避免强依赖关系。增删一个模块，应该只会影响有依赖关系的相关模块，无关的不应该受影响。

所谓低耦合：从小的角度来看，就是要每个 Java 类之间的耦合性降低，多用接口，利用 Java 面向对象编程思想的封装、继承、多态，隐藏实现细节。从模块之间来讲，就是要每个模块之间的关系降低，减少冗余、重复、交叉的复杂度，模块功能划分尽可能单一。

在音视频应用技术中，我们知道其实主要占用的资源是 cpu、memory，而且涉及到资源的共享问题，所以需要结合 NFS 来实现跨节点的资源共享。当然，单节点暴露的问题是，如果一旦客户端与服务器保持长时间的连接，而且，不同客户端同时发送请求，此时，单节点的压力是很大的。很有可能导致 cpu、memory 吃紧，从而导致节点的 crash，这样，不利于系统的高可用、服务的健壮性。此时，需要解决的是音视频通信中的资源吃紧的问题，在系统领域，通常可以采用多节点的方式，来实现分布式、高并发请求，当请求过来时，可以通过负载均衡的方式，通过一定的策略，如：根据最小请求数，或为每一个服务器赋予一个权重值，服务器响应时间越长，这个服务器的权重就越小，被选中的几率就会降低。这样来控制服务请求压力，从而让客户端与服务器能够保持长时间、有效的进行通信。

如何使用 Springboot 框架搭建微服务

介绍

这几年的快速发展，微服务已经变得越来越流行。其中，Spring Cloud 一直在更新，并被大部分公司所使用。代表性的有 Alibaba，2018 年 11 月左右，Spring Cloud 联合创始人 Spencer Gibb 在 Spring 官网的博客页面宣布：阿里巴巴开源 Spring Cloud Alibaba，并发布了首个预览版本。随后，Spring Cloud 官方 Twitter 也发布了此消息。

在 Spring Boot1.x 中，主要包括 Eureka、Zuul、Config、Ribbon、Hystrix 等。而在 Spring Boot2.x 中，网关采用了自己的 Gateway。当然在 Alibaba 版本中，其组件更是丰富：使用 Alibaba 的 Nacos 作为注册中心和配置中心。使用自带组件 Sentinel 作为限流、熔断神器。

搭建注册中心

我们今天主要来利用 Springboot 结合阿里巴巴的插件来实现微聊天系统的微服务设计。首先先来创建一个注册中心 Nacos。

我们先下载 Nacos，Nacos 地址：github.com/alibaba/nac…我们下载对应系统的二进制文件后，对应自己的系统，执行如下命令：

1
2
3
复制代码Linux/Unix/Mac：sh startup.sh -m standalone

Windows：cmd startup.cmd -m standalone

启动完成之后，访问：http://127.0.0.1:8848/nacos/，可以进入 Nacos 的服务管理页面，具体如下：

img

默认用户名与密码都是 nacos。

登陆后打开服务管理，可以看到注册到 Nacos 的服务列表：

img

可以点击配置管理，查看配置：

img

如果没有配置任何服务的配置，可以新建：

img

上面讲述了 Nacos 如何作为注册中心与配置中心的，很简单吧。

第一个微服务

接下来，对于微服务，那需要有一个服务被注册与被发现，我们讲解服务提供者代码：

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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
xml复制代码<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.damon</groupId>
  <artifactId>provider-service</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>


  <name>provider-service</name>
  <url>http://maven.apache.org</url>


  <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.8.RELEASE</version>
        <relativePath/>
    </parent>


  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
    <swagger.version>2.6.1</swagger.version>
    <xstream.version>1.4.7</xstream.version>
    <pageHelper.version>4.1.6</pageHelper.version>
    <fastjson.version>1.2.51</fastjson.version>
    <!-- <springcloud.version>2.1.8.RELEASE</springcloud.version> -->
    <springcloud.version>Greenwich.SR3</springcloud.version>
    <springcloud.kubernetes.version>1.1.1.RELEASE</springcloud.kubernetes.version>
    <mysql.version>5.1.46</mysql.version>

    <alibaba-cloud.version>2.1.1.RELEASE</alibaba-cloud.version>
    <springcloud.alibaba.version>0.9.0.RELEASE</springcloud.alibaba.version>
  </properties>


  <dependencyManagement>
    <dependencies>
      <!-- <dependency>
          <groupId>com.alibaba.cloud</groupId>
          <artifactId>spring-cloud-alibaba-dependencies</artifactId>
          <version>${alibaba-cloud.version}</version>
          <type>pom</type>
          <scope>import</scope>
      </dependency> -->


      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-alibaba-dependencies</artifactId>
        <version>${springcloud.alibaba.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>

      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-dependencies</artifactId>
        <version>${springcloud.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
  </dependencyManagement>


  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
      <exclusions>
        <exclusion>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-tomcat</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-undertow</artifactId>
    </dependency>

    <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-oauth2</artifactId>
        </dependency>

        <dependency>
          <groupId>org.springframework.cloud</groupId>
          <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
      </dependency>

      <dependency>
          <groupId>org.springframework.cloud</groupId>
          <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
      </dependency>

      <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
        </dependency>


        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
        </dependency>

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

    <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>${fastjson.version}</version>
    </dependency>

    <!-- swagger -->
    <dependency>
      <groupId>io.springfox</groupId>
      <artifactId>springfox-swagger2</artifactId>
      <version>${swagger.version}</version>
    </dependency>
    <dependency>
      <groupId>io.springfox</groupId>
      <artifactId>springfox-swagger-ui</artifactId>
      <version>${swagger.version}</version>
    </dependency>


    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
        </dependency>

    <dependency>
        <groupId>commons-collections</groupId>
        <artifactId>commons-collections</artifactId>
        <version>3.2.2</version>
    </dependency>


    <!--分页插件-->
      <dependency>
        <groupId>com.github.pagehelper</groupId>
        <artifactId>pagehelper</artifactId>
        <version>${pageHelper.version}</version>
      </dependency>

    <!-- mybatis -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.1.1</version>
        </dependency>
        <dependency>
      <groupId>mysql</groupId>
      <artifactId>mysql-connector-java</artifactId>
      <version>${mysql.version}</version>
    </dependency>

        <!-- datasource pool-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.1.3</version>
        </dependency>

        <!-- 对redis支持,引入的话项目缓存就支持redis了,所以必须加上redis的相关配置,否则操作相关缓存会报异常 -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

    <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>19.0</version>
        </dependency>

        <dependency>
      <groupId>io.jsonwebtoken</groupId>
      <artifactId>jjwt</artifactId>
      <version>0.9.0</version>
    </dependency>

  </dependencies>


  <build>
    <finalName>${project.artifactId}</finalName>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
        <configuration>
          <jvmArguments>-Dfile.encoding=UTF-8</jvmArguments>
          <fork>true</fork>
        </configuration>
      </plugin>
      <plugin>
        <groupId>org.jacoco</groupId>
        <artifactId>jacoco-maven-plugin</artifactId>
        <version>0.7.8</version>
        <executions>
          <execution>
            <goals>
              <goal>prepare-agent</goal>
              <goal>report</goal>
            </goals>
          </execution>
        </executions>
      </plugin>

      <!-- 自动生成代码 插件 begin -->
      <!-- <plugin>
        <groupId>org.mybatis.generator</groupId>
        <artifactId>mybatis-generator-maven-plugin</artifactId>
        <version>1.3.2</version>
        <configuration>
          <configurationFile>src/main/resources/generatorConfig.xml</configurationFile>
          <verbose>true</verbose>
          <overwrite>true</overwrite>
        </configuration>
        <dependencies>
          <dependency>
            <groupId>org.mybatis.generator</groupId>
            <artifactId>mybatis-generator-core</artifactId>
            <version>1.3.2</version>
          </dependency>
        </dependencies>
      </plugin> -->
    </plugins>
  </build>
</project>

一如既往的引入依赖，配置 bootstrap 文件:

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
yaml复制代码management:
  endpoint:
    restart:
      enabled: true
    health:
      enabled: true
    info:
      enabled: true

spring:
  application:
    name: provider-service

  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
      config:
        server-addr: 127.0.0.1:8848
        refreshable-dataids: actuator.properties,log.properties

  http:
    encoding:
      charset: UTF-8
      enabled: true
      force: true
  mvc:
    throw-exception-if-no-handler-found: true
  main:
    allow-bean-definition-overriding: true #当遇到同样名称时，是否允许覆盖注册



logging:
  path: /data/${spring.application.name}/logs


cas-server-url: http://oauth-cas #http://localhost:2000#设置可以访问的地址


security:
  oauth2: #与cas对应的配置
    client:
      client-id: provider-service
      client-secret: provider-service-123
      user-authorization-uri: ${cas-server-url}/oauth/authorize #是授权码认证方式需要的
      access-token-uri: ${cas-server-url}/oauth/token #是密码模式需要用到的获取 token 的接口
    resource:
      loadBalanced: true
      #jwt: #jwt存储token时开启
        #key-uri: ${cas-server-url}/oauth/token_key
        #key-value: test_jwt_sign_key
      id: provider-service
      #指定用户信息地址
      user-info-uri: ${cas-server-url}/api/user #指定user info的URI，原生地址后缀为/auth/user
      prefer-token-info: false
      #token-info-uri:
    authorization:
      check-token-access: ${cas-server-url}/oauth/check_token #当此web服务端接收到来自UI客户端的请求后，需要拿着请求中的 token 到认证服务端做 token 验证，就是请求的这个接口
application 文件；

server:
  port: 2001
  undertow:
    accesslog:
      enabled: false
      pattern: combined
  servlet:
    session:
      timeout: PT120M
      cookie:
        name: PROVIDER-SERVICE-SESSIONID #防止Cookie冲突，冲突会导致登录验证不通过



client:
  http:
    request:
      connectTimeout: 8000
      readTimeout: 30000

mybatis:
  mapperLocations: classpath:mapper/*.xml
  typeAliasesPackage: com.damon.*.model



backend:
  ribbon:
    client:
      enabled: true
    ServerListRefreshInterval: 5000


ribbon:
  ConnectTimeout: 3000
  # 设置全局默认的ribbon的读超时
  ReadTimeout: 1000
  eager-load:
    enabled: true
    clients: oauth-cas,consumer-service
  MaxAutoRetries: 1 #对第一次请求的服务的重试次数
  MaxAutoRetriesNextServer: 1 #要重试的下一个服务的最大数量（不包括第一个服务）
  #listOfServers: localhost:5556,localhost:5557
  #ServerListRefreshInterval: 2000
  OkToRetryOnAllOperations: true
  NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RoundRobinRule




hystrix.command.BackendCall.execution.isolation.thread.timeoutInMilliseconds: 5000
hystrix.threadpool.BackendCallThread.coreSize: 5

接下来启动类：

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
kotlin复制代码package com.damon;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.security.oauth2.client.EnableOAuth2Sso;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;




/**
 * @author Damon
 * @date 2020年1月13日 下午3:23:06
 *
 */


@Configuration
@EnableAutoConfiguration
@ComponentScan(basePackages = {"com.damon"})
@EnableDiscoveryClient
@EnableOAuth2Sso
public class ProviderApp {


    public static void main(String[] args) {
        SpringApplication.run(ProviderApp.class, args);
    }


}

注意：注解 @EnableDiscoveryClient、@EnableOAuth2Sso 都需要。

这时，同样需要配置 ResourceServerConfig、SecurityConfig。

如果需要数据库，可以加上：

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
java复制代码package com.damon.config;

import java.util.Properties;
import javax.sql.DataSource;
import org.apache.ibatis.plugin.Interceptor;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.EnableTransactionManagement;


import com.alibaba.druid.pool.DruidDataSourceFactory;
import com.github.pagehelper.PageHelper;


/**
*
*
* created by Damon
* 2018年5月23日 下午7:39:37
*
*/
@Component
@Configuration
@EnableTransactionManagement
@MapperScan("com.damon.*.dao")
public class MybaitsConfig {


    @Autowired
    private EnvConfig envConfig;

    @Autowired
  private Environment env;


    @Bean(name = "dataSource")
    public DataSource getDataSource() throws Exception {
        Properties props = new Properties();
        props.put("driverClassName", envConfig.getJdbc_driverClassName());
        props.put("url", envConfig.getJdbc_url());
        props.put("username", envConfig.getJdbc_username());
        props.put("password", envConfig.getJdbc_password());
        return DruidDataSourceFactory.createDataSource(props);
    }


    @Bean
  public SqlSessionFactory sqlSessionFactory(@Qualifier("dataSource") DataSource dataSource) throws Exception {


    SqlSessionFactoryBean fb = new SqlSessionFactoryBean();
    // 指定数据源(这个必须有，否则报错)
    fb.setDataSource(dataSource);
    // 下边两句仅仅用于*.xml文件，如果整个持久层操作不需要使用到xml文件的话（只用注解就可以搞定），则不加
    fb.setTypeAliasesPackage(env.getProperty("mybatis.typeAliasesPackage"));// 指定基包
    fb.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(env.getProperty("mybatis.mapperLocations")));// 指定xml文件位置


    // 分页插件
    PageHelper pageHelper = new PageHelper();
    Properties props = new Properties();
    // 启用合理化时，如果pageNum<1会查询第一页，如果pageNum>pages会查询最后一页
        //禁用合理化时，如果pageNum<1或pageNum>pages会返回空数据
    props.setProperty("reasonable", "true");
    //指定数据库
    props.setProperty("dialect", "mysql");
    //支持通过Mapper接口参数来传递分页参数
    props.setProperty("supportMethodsArguments", "true");
    //总是返回PageInfo类型,check检查返回类型是否为PageInfo,none返回Page
    props.setProperty("returnPageInfo", "check");
    props.setProperty("params", "count=countSql");
    pageHelper.setProperties(props);
    // 添加插件
    fb.setPlugins(new Interceptor[] { pageHelper });


    try {
      return fb.getObject();
    } catch (Exception e) {
            throw e;
    }
  }


    /**
     * 配置事务管理器
     * @param dataSource
     * @return
     * @throws Exception
     */
    @Bean
    public DataSourceTransactionManager transactionManager(DataSource dataSource) throws Exception {
        return new DataSourceTransactionManager(dataSource);
    }

    @Bean
    public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {
        return new SqlSessionTemplate(sqlSessionFactory);
    }


}

接下来新写一个 controller 类：

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
kotlin复制代码package com.damon.user.controller;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.Authentication;
import org.springframework.security.oauth2.provider.authentication.OAuth2AuthenticationDetails;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


import com.damon.commons.Response;
import com.damon.user.service.UserService;


/**
 *
 *
 * @author Damon
 * @date 2020年1月13日 下午3:31:07
 *
 */
@RestController
@RequestMapping("/api/user")
public class UserController {

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

  @Autowired
  private UserService userService;

    @GetMapping("/getCurrentUser")
    @PreAuthorize("hasAuthority('admin')")
    public Object getCurrentUser(Authentication authentication) {
      logger.info("test password mode");
        return authentication;
    }


    @PreAuthorize("hasAuthority('admin')")
    @GetMapping("/auth/admin")
    public Object adminAuth() {
      logger.info("test password mode");
        return "Has admin auth!";
    }

    @GetMapping(value = "/get")
    @PreAuthorize("hasAuthority('admin')")
    //@PreAuthorize("hasRole('admin')")//无效
    public Object get(Authentication authentication){
        //Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        authentication.getCredentials();
        OAuth2AuthenticationDetails details = (OAuth2AuthenticationDetails)authentication.getDetails();
        String token = details.getTokenValue();
        return token;
    }

    @GetMapping("/getUserInfo")
    @PreAuthorize("hasAuthority('admin')")
    public Response<Object> getUserInfo(Authentication authentication) {
      logger.info("test password mode");
      Object principal = authentication.getPrincipal();
      if(principal instanceof String) {
        String username = (String) principal;
        return userService.getUserByUsername(username);
      }
    return null;
    }


}

基本上一个代码就完成了。接下来测试一下：

认证：

1
bash复制代码curl -i -X POST -d "username=admin&password=123456&grant_type=password&client_id=provider-service&client_secret=provider-service-123" http://localhost:5555/oauth-cas/oauth/token

拿到 token 后：

1
bash复制代码curl -i -H "Accept: application/json" -H "Authorization:bearer f4a42baa-a24a-4342-a00b-32cb135afce9" -X GET http://localhost:5555/provider-service/api/user/getCurrentUser

这里用到了 5555 端口，这是一个网关服务，好吧，既然提到这个，我们接下来看网关吧，引入依赖：

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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
xml复制代码<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.damon</groupId>
  <artifactId>alibaba-gateway</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>


  <name>alibaba-gateway</name>
  <url>http://maven.apache.org</url>


  <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.8.RELEASE</version>
        <relativePath/>
    </parent>


  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
    <swagger.version>2.6.1</swagger.version>
    <xstream.version>1.4.7</xstream.version>
    <pageHelper.version>4.1.6</pageHelper.version>
    <fastjson.version>1.2.51</fastjson.version>
    <!-- <springcloud.version>2.1.8.RELEASE</springcloud.version> -->
    <springcloud.version>Greenwich.SR3</springcloud.version>
    <springcloud.kubernetes.version>1.1.1.RELEASE</springcloud.kubernetes.version>
    <mysql.version>5.1.46</mysql.version>

    <alibaba-cloud.version>2.1.1.RELEASE</alibaba-cloud.version>
    <springcloud.alibaba.version>0.9.0.RELEASE</springcloud.alibaba.version>
  </properties>


  <dependencyManagement>
    <dependencies>
      <!-- <dependency>
          <groupId>com.alibaba.cloud</groupId>
          <artifactId>spring-cloud-alibaba-dependencies</artifactId>
          <version>${alibaba-cloud.version}</version>
          <type>pom</type>
          <scope>import</scope>
      </dependency> -->


      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-alibaba-dependencies</artifactId>
        <version>${springcloud.alibaba.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>

      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-dependencies</artifactId>
        <version>${springcloud.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
  </dependencyManagement>


  <dependencies>
    <!-- 不要依赖spring-boot-starter-web，会和spring-cloud-starter-gateway冲突，启动时异常 -->
    <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-gateway</artifactId>
        </dependency>

        <!--基于 reactive stream 的redis -->
    <!-- <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
    </dependency> -->

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-commons</artifactId>
        </dependency>


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

        <dependency>
          <groupId>org.springframework.cloud</groupId>
          <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
      </dependency>


        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
        </dependency>


        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
        </dependency>

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

        <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>${fastjson.version}</version>
    </dependency>

    <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>19.0</version>
        </dependency>

    </dependencies>

    <build>
    <finalName>${project.artifactId}</finalName>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
        <configuration>
          <jvmArguments>-Dfile.encoding=UTF-8</jvmArguments>
          <fork>true</fork>
        </configuration>
      </plugin>
      <plugin>
        <groupId>org.jacoco</groupId>
        <artifactId>jacoco-maven-plugin</artifactId>
        <version>0.7.8</version>
        <executions>
          <execution>
            <goals>
              <goal>prepare-agent</goal>
              <goal>report</goal>
            </goals>
          </execution>
        </executions>
      </plugin>

      <!-- 自动生成代码 插件 begin -->
      <!-- <plugin>
        <groupId>org.mybatis.generator</groupId>
        <artifactId>mybatis-generator-maven-plugin</artifactId>
        <version>1.3.2</version>
        <configuration>
          <configurationFile>src/main/resources/generatorConfig.xml</configurationFile>
          <verbose>true</verbose>
          <overwrite>true</overwrite>
        </configuration>
        <dependencies>
          <dependency>
            <groupId>org.mybatis.generator</groupId>
            <artifactId>mybatis-generator-core</artifactId>
            <version>1.3.2</version>
          </dependency>
        </dependencies>
      </plugin> -->
    </plugins>
  </build>
</project>

同样利用 Nacos 来发现服务。

这里的注册配置为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
yaml复制代码spring:
  cloud:
    gateway:
      discovery:
        locator:
          enabled: true #并且我们并没有给每一个服务单独配置路由 而是使用了服务发现自动注册路由的方式
          lowerCaseServiceId: true

    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
      config:
        server-addr: 127.0.0.1:8848
        refreshable-dataids: actuator.properties,log.properties

前面用的是 kubernetes。

好了，网关配置好后，启动在 Nacos dashboard 可以看到该服务，表示注册服务成功。接下来就可以利用其来调用其他服务了。具体 curl 命令：

1
bash复制代码curl -i -H "Accept: application/json" -H "Authorization:bearer f4a42baa-a24a-4342-a00b-32cb135afce9" -X GET http://localhost:5555/consumer-service/api/order/getUserInfo

Ok，到此鉴权中心、服务提供者、服务消费者、服务的注册与发现、配置中心等功能已完成。

为什么选择 Netty 作为即时通信的技术框架

简介

Netty 是一个高性能、异步事件驱动的 NIO 框架，它提供了对 TCP、UDP 和文件传输的支持。作为当前最流行的 NIO 框架，Netty 在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用。

特点

  • 高并发
  • 传输快
  • 封装好

Netty 通信的优势

Netty 是一个高性能、高可拓展性的异步事件驱动的网络应用程序框架，极大地简化了 TCP 和 UDP 客户端和服务器端开发等网络编程，它的四个重要内容：

  • 内存管理：增强 ByteBuf 缓冲区
  • Reactor 线程模型：一种高性能的多线程程序设计
  • 增强版的通道 channel 概念
  • ChannelPipeline 责任链设计模式：事件处理机制

Netty 实现了 Reactor 线程模型，Reactor 模型有四个核心概念：Resources 资源（请求/任务）、Synchronous Event Demultiplexer 同步事件复用器、Dispatcher 分配器、Request Handler 请求处理器。主要是通过 2 个 EventLoopGroup（线程组，底层是 JDK 的线程池）来分别处理连接和数据读取，从而提高线程的利用率。

Netty 中的 Channel 是一个抽象的概念，可以理解为对 JDK NIO Channel 的增强和拓展。增加了很多属性和方法。

ChannelPipeline 责任链保存了通道所有处理器信息。创建新 channel 时自动创建一个专有的 pipeline，并且在对应入站事件（通常指 I/O 线程生成了入站数据，详见 ChannelInboundHandler）和出站事件（经常是指 I/O 线程执行实际的输出操作，详见 ChannelOutboundHandler）时调用 pipeline 上的处理器。当入站事件时，执行顺序是 pipeline 的 first 执行到 last。当出站事件时，执行顺序是 pipeline 的 last 执行到 first。处理器在 pipeline 中的顺序由添加的时候决定。

JDK 的 ByteBuffer 存在如无法动态扩容、API 使用复杂的问题，Netty 自己的 ByteBuf 解决了其问题。ByteBuf 实现了四个方面的增强：API 操作便捷，动态扩容，多种 ByteBuf 实现，高效的零拷贝机制。

实现一个简单的 Netty 客户端、服务器通信

实战服务端

前面介绍了 Netty 在音视频流域实践的优势与特点，接下来，我们先写一个服务端。首先创建一个 Java 项目：

img

创建项目后，我们需要引入基础依赖：

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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
xml复制代码<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.damon</groupId>
  <artifactId>netty-client-service</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>netty-client-service</name>
  <url>http://maven.apache.org</url>

  <parent>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-parent</artifactId>
       <version>2.1.1.RELEASE</version>
       <relativePath/>
   </parent>

  <properties>
        <java.version>1.8</java.version>
        <spring-boot.version>2.1.1.RELEASE</spring-boot.version>
        <springcloud.kubernetes.version>1.0.1.RELEASE</springcloud.kubernetes.version>
        <springcloud.version>2.1.1.RELEASE</springcloud.version>
        <swagger.version>2.6.1</swagger.version>
		<fastjson.version>1.2.51</fastjson.version>
		<pageHelper.version>4.1.6</pageHelper.version>
		<protostuff.version>1.0.10</protostuff.version>
        <objenesis.version>2.4</objenesis.version>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <type>pom</type>
                <scope>import</scope>
                <version>${spring-boot.version}</version>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
    
    	<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
			<exclusions>
				<exclusion>
					<groupId>org.springframework.boot</groupId>
					<artifactId>spring-boot-starter-tomcat</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-undertow</artifactId>
		</dependency>
		
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
    
        <!-- <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-kubernetes-core</artifactId>
            <version>${springcloud.kubernetes.version}</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-kubernetes-discovery</artifactId>
            <version>${springcloud.kubernetes.version}</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-kubernetes-ribbon</artifactId>
            <version>${springcloud.kubernetes.version}</version>
        </dependency> -->

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-commons</artifactId>
            <version>${springcloud.version}</version>
        </dependency>

		<dependency>
			<groupId>com.alibaba</groupId>
			<artifactId>fastjson</artifactId>
			<version>${fastjson.version}</version>
		</dependency>
		
		<dependency>
		  <groupId>org.jsoup</groupId>
		  <artifactId>jsoup</artifactId>
		  <version>1.11.3</version>
		</dependency>

		<!-- swagger -->
		<dependency>
			<groupId>io.springfox</groupId>
			<artifactId>springfox-swagger2</artifactId>
			<version>${swagger.version}</version>
		</dependency>
		<dependency>
			<groupId>io.springfox</groupId>
			<artifactId>springfox-swagger-ui</artifactId>
			<version>${swagger.version}</version>
		</dependency>

		<dependency>
		    <groupId>org.apache.commons</groupId>
		    <artifactId>commons-lang3</artifactId>
		</dependency>
		
		<dependency>
		    <groupId>commons-collections</groupId>
		    <artifactId>commons-collections</artifactId>
		    <version>3.2.2</version>
		</dependency>
		
		<!-- mybatis -->
        <!-- <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.1.1</version>
        </dependency> -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        
        <!-- <dependency>
            <groupId>org.bytedeco</groupId>
            <artifactId>javacv-platform</artifactId>
            <version>1.4.1</version>
        </dependency> -->
        <!-- <dependency>
            <groupId>org.bytedeco.javacpp-presets</groupId>
            <artifactId>opencv-platform</artifactId>
            <version>3.4.1-1.4.1</version>
        </dependency> -->
        
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.64.Final</version>
        </dependency>
        
        <!-- protobuf -->
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>3.5.0</version>
        </dependency>
        <dependency>
            <groupId>com.googlecode.protobuf-java-format</groupId>
            <artifactId>protobuf-java-format</artifactId>
            <version>1.2</version>
        </dependency>
        
        <dependency>
            <groupId>com.dyuproject.protostuff</groupId>
            <artifactId>protostuff-core</artifactId>
            <version>${protostuff.version}</version>
        </dependency>
        <dependency>
            <groupId>com.dyuproject.protostuff</groupId>
            <artifactId>protostuff-runtime</artifactId>
            <version>${protostuff.version}</version>
        </dependency>
        <dependency>
            <groupId>org.objenesis</groupId>
            <artifactId>objenesis</artifactId>
            <version>${objenesis.version}</version>
        </dependency>
		
	</dependencies>
  </project>

服务启动类：

1
2
3
4
5
6
7
8
9
less复制代码@EnableScheduling
@SpringBootApplication(scanBasePackages = { "com.damon" })
public class StorageServer {
	
	public static void main(String[] args) {
		SpringApplication.run(StorageServer.class, args);
	}

}

首先启动 netty 服务时，只需要我们添加 Netty 的配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
ini复制代码spring.application.name=netty-server
server.port=2002
netty.host=127.0.0.1
netty.port=9999

logging.path=/data/${spring.application.name}/logs
spring.profiles.active=dev

spring.http.encoding.charset=UTF-8
spring.http.encoding.enabled=true 
spring.http.encoding.force=true
spring.mvc.throw-exception-if-no-handler-found=true

server.undertow.accesslog.enabled=false
server.undertow.accesslog.pattern=combined

client.http.request.readTimeout=30000
client.http.request.connectTimeout=8000

添加完配置，我们可以启动服务看看，这时候有日志：

img

添加完 netty 服务配置后，这里需要注入一个 Server Handle，用来当客户端主动链接服务端的链接后，这时候，该处理类会被触发，从而执行一些消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
ini复制代码@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		SocketChannel channel = (SocketChannel) ctx.channel();
		logger.info("链接报告开始");
		logger.info("链接报告信息：有一客户端链接到本服务端");
		logger.info("链接报告IP:{}", channel.localAddress().getHostString());
		logger.info("链接报告Port:{}", channel.localAddress().getPort());
		logger.info("链接报告完毕");
		ChannelHandler.channelGroup.add(ctx.channel());
		// 通知客户端链接建立成功
		String str = "通知客户端链接建立成功" + " " + new Date() + " " + channel.localAddress().getHostString() + "\r\n";
		ByteBuf buf = Unpooled.buffer(str.getBytes().length);
        buf.writeBytes(str.getBytes("GBK"));
        ctx.writeAndFlush(buf);
	}

意思就是说，假如这时候有个客户端连接服务端时，会被打印一些信息，这里是我提前加入客户端后打印的结果：

img

当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据：

1
2
3
4
5
less复制代码@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		logger.info("客户端断开链接{}", ctx.channel().localAddress().toString());
		ChannelHandler.channelGroup.remove(ctx.channel());
	}

当然获取数据函数在这里：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
ini复制代码@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws UnsupportedEncodingException {
		
		if(msg instanceof ByteBuf) {
			ByteBuf buf = (ByteBuf) msg;
	        byte[] msgByte = new byte[buf.readableBytes()];
	        buf.readBytes(msgByte);
	        System.out.println(new String(msgByte, Charset.forName("GBK")));
			
	        //通知客户端链消息发送成功
	        String str = "服务端收到：" + new Date() + " " + new String(msgByte, Charset.forName("GBK")) + "\r\n";
	        ByteBuf buf2 = Unpooled.buffer(str.getBytes().length);
	        buf2.writeBytes(str.getBytes("GBK"));
	        ctx.writeAndFlush(buf2);
		}
  }

如果出现异常，抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接**：**

1
2
3
4
5
java复制代码@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		ctx.close();
		logger.info("异常信息：\r\n" + cause.getMessage());
	}

此外，在服务端，一般需要定义一些信息协议信息，如：连接的信息，是自发信息还是群发信息，通信管道是哪个，还有通信信息等：

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
typescript复制代码public class ServerMsgProtocol {

    private int type;             //链接信息;1自发信息、2群发消息
    private String channelId;     //通信管道ID，实际使用中会映射成用户名
    private String userHeadImg;   //用户头像[模拟分配]
    private String msgInfo;       //通信消息

    public int getType() {
        return type;
    }

    public void setType(int type) {
        this.type = type;
    }

    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public String getUserHeadImg() {
        return userHeadImg;
    }

    public void setUserHeadImg(String userHeadImg) {
        this.userHeadImg = userHeadImg;
    }

    public String getMsgInfo() {
        return msgInfo;
    }

    public void setMsgInfo(String msgInfo) {
        this.msgInfo = msgInfo;
    }
}

以上，就是一个简单的服务端，梳理一下还是比较清晰的。

实战客户端

接下来，我们看看客户端是如何连接服务端，并且与其通信的呢？客户端要想与服务端通信，首先肯定需要与服务端进行连接，这里加一个配置服务端 NIO 线程组：

1
2
ini复制代码private EventLoopGroup workerGroup = new NioEventLoopGroup();
private Channel channel;

连接服务端的逻辑是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
ini复制代码public ChannelFuture connect(String inetHost, int inetPort) {
        ChannelFuture channelFuture = null;
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.AUTO_READ, true);
            b.handler(new MyChannelInitializer());
            channelFuture = b.connect(inetHost, inetPort).syncUninterruptibly();
            this.channel = channelFuture.channel();
            channel.closeFuture();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != channelFuture && channelFuture.isSuccess()) {
                System.out.println("demo-netty client start done.");
            } else {
                System.out.println("demo-netty client start error.");
            }
        }
        return channelFuture;
}

接下来再看如何销毁连接：

1
2
3
4
5
csharp复制代码public void destroy() {
	if (null == channel) return;
	channel.close();
	workerGroup.shutdownGracefully();
}

最后，我们来连接到服务端：

1
scss复制代码new NettyClient().connect("127.0.0.1", 9999);

由于前面我们的服务端的 netty 的 ip 与端口设置为：本地，9999 端口，这里直接配置。

同样的，客户端如果需要接收数据信息，也需要定义如何在管道中进行接收：

1
2
3
4
5
6
7
8
9
scala复制代码public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        // 在管道中添加我们自己的接收数据实现方法
        channel.pipeline().addLast(new MyClientHandler());
    }

}

当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据：

1
2
3
4
5
6
7
8
9
csharp复制代码@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SocketChannel channel = (SocketChannel) ctx.channel();
        System.out.println("链接报告开始");
        System.out.println("链接报告信息：本客户端链接到服务端。channelId：" + channel.id());
        System.out.println("链接报告IP:" + channel.localAddress().getHostString());
        System.out.println("链接报告Port:" + channel.localAddress().getPort());
        System.out.println("链接报告完毕");
    }

当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据：

1
2
3
4
5
java复制代码@Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("断开链接" + ctx.channel().localAddress().toString());
        super.channelInactive(ctx);
    }

遇到异常时，抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接：

1
2
3
4
5
java复制代码@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
	ctx.close();
	System.out.println("异常信息：\r\n" + cause.getMessage());
}

客户端连接服务端、处理接收服务端发送的信息、异常处理等完成后，这时候，我们来启动客户端，客户端控制面会打印如下信息：

img

如果客户端主动断开连接时，这时候，服务端会提示：

1
2
yaml复制代码远程主机强迫关闭了一个现有的连接。
2021-05-13 19:33:35.691  INFO 148736 --- [ntLoopGroup-3-2] com.leinao.handler.ServerHandler         : 客户端断开链接/127.0.0.1:9999

到此，一个简单的 Netty 客户端、服务端的通信就完成了。

微服务 Springboot 下实战聊天系统

在前面介绍了一个简单的 Netty 客户端、服务端通信的示例，接下来，我们开始实战聊天系统。

websocket 服务端启动类

基于前面讲的 Netty 的特性，这里聊天室需要前、后端。那么，首先对于后端，我们需要创建一个 Websocket Server，这里需要有一对线程组 EventLoopGroup，定义完后，需要定义一个 Server：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
scss复制代码public static void main(String[] args) throws Exception {
  
  EventLoopGroup mainGroup = new NioEventLoopGroup();
  EventLoopGroup subGroup = new NioEventLoopGroup();
  
  try {
   ServerBootstrap server = new ServerBootstrap();
   server.group(mainGroup, subGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new WSServerInitialzer());
   
   ChannelFuture future = server.bind(8088).sync();
   
   future.channel().closeFuture().sync();
  } finally {
   mainGroup.shutdownGracefully();
   subGroup.shutdownGracefully();
  }
}

将线程组加入 Server，接下来，需要设置一个 channel：NioServerSocketChannel，还有一个初始化器：WSServerInitialzer。

第二步，需要对 Server 进行端口版绑定：

1
ini复制代码ChannelFuture future = server.bind(8088).sync()

最后，需要对 future 进行监听。而且监听结束后需要对线程资源进行关闭：

1
2
ini复制代码mainGroup.shutdownGracefully();
subGroup.shutdownGracefully();

websocket 子处理器 initialzer

上面说了 WebSocket Server，那么对于 socket，有一个初始化处理器，这里我们来定义一个：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
scala复制代码public class WSServerInitialzer extends ChannelInitializer<SocketChannel> {

 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
  ChannelPipeline pipeline = ch.pipeline();
  pipeline.addLast(new HttpServerCodec());
  pipeline.addLast(new ChunkedWriteHandler());
  pipeline.addLast(new HttpObjectAggregator(1024*64));

  pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
  
  pipeline.addLast(new ChatHandler());
 }

}

由于 websocket 是基于 http 协议，所以需要有 http 的编解码器 HttpServerCodec，同时，在一些 http 上，有一些数据流的处理，而且，数据流有大有小，那么可以添加一个大数据流的处理：ChunkedWriteHandler。

通常，会有对 httpMessage 进行聚合，聚合成 FullHttpRequest 或 FullHttpResponse，而且，几乎在 netty 中的编程，都会使用到此 hanler。

另外，websocket 服务器处理的协议，用于指定给客户端连接访问的路由 : “/ws”，本 handler 会帮你处理一些繁重的复杂的事，比如，会帮你处理握手动作： handshaking（close, ping, pong） ping + pong = 心跳。对于 websocket 来讲，都是以 frames 进行传输的，不同的数据类型对应的 frames 也不同。

最后，我们自定义了一个处理消息的 handler：ChatHandler。

chatHandler 对消息的处理

在 Netty 中，有一个用于为 websocket 专门处理文本的对象 TextWebSocketFrame，frame 是消息的载体。

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
java复制代码public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
 private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 
 @Override
 protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) 
   throws Exception {
  String content = msg.text();
  System.out.println("接受到的数据：" + content);
  
  clients.writeAndFlush(new TextWebSocketFrame("服务器时间在 " + LocalDateTime.now() + " 接受到消息, 消息为：" + content));

 }

 @Override
 public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  clients.add(ctx.channel());
  System.out.println("客户端连接，channle对应的长id为：" + ctx.channel().id().asLongText());
  System.out.println("客户端连接，channle对应的短id为：" + ctx.channel().id().asShortText());
 }

 @Override
 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  System.out.println("客户端断开，channle对应的长id为：" + ctx.channel().id().asLongText());
  System.out.println("客户端断开，channle对应的短id为：" + ctx.channel().id().asShortText());
 }
}

一开始消息在载体 TextWebSocketFrame 中，这时候可以直接拿到其中的内容，并且打印出来。而且可以把消息发到对应请求的客户端。当然，也可以把消息转发给所有的客户端，这就涉及到 Netty 中的 channel。这时候，需要管理 channel 中的用户，这样才能把消息转发到所有 channel 的用户。也就是上面的 handlerAdded 函数，当客户端连接服务端之后打开连接，获取客户端的 channle，并且放到 ChannelGroup 中去进行管理。同时，客户端与服务端断开、关闭连接后，会触发 handlerRemoved 函数，同时 ChannelGroup 会自动移除对应客户端的 channel。

接下来，需要把数据获取后刷新到所有客户端：

1
2
3
less复制代码for (Channel channel : clients) {
   channel.writeAndFlush(new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() + "接受到消息, 消息为：" + content));
}

注意：这里需要借助于载体来把信息 Flush，因为 writeAndFlush 函数是需要传对象载体，而不是直接字符串。其实同样，作为 ChannelGroup clients，其本身提供了 writeAndFlush 函数，可以直接输出到所有客户端：

1
less复制代码clients.writeAndFlush(new TextWebSocketFrame("服务器时间在 " + LocalDateTime.now() + " 接受到消息, 消息为：" + content));

基于 js 的 websocket 相关 api 介绍

首先，需要一个客户端与服务端的连接，这个连接桥梁在 js 中就是一个 socket：

1
ini复制代码var socket = new WebSocket("ws://192.168.174.145:8088/ws");

再来看看其生命周期，在后端，channel 有其生命周期，而前端 socket 中:

  • onopen()，当客户端与服务端建立连接时，就会触发 onopen 事件
  • onmessage()，是在客户端收到消息时，就会触发 onmessage 事件
  • onerror()，出现异常时，前端会触发 onerror 事件
  • onclose()，客户端与服务端连接关闭后，就会触发 onclose 事件

接下来看看两个主动的方法：

  • Socket.send()，在前端主动获取内容后，通过 send 进行消息发送
  • Socket.close()，当用户触发某个按钮，就会断开客户端与服务端的连接

以上就是对于前端 websocket js 相对应的 api。

实现前端 websocket

上面介绍了后端对于消息的处理、编解码等，又介绍了 websocket js 的相关。接下来，我们看看前端如何实现 websocket，首先我们先写一个文本输入、点击等功能:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
xml复制代码<html>
 <head>
  <meta charset="utf-8" />
  <title></title>
 </head>
 <body>
  
  <div>send msg:</div>
  <input type="text" id="msgContent"/>
  <input type="button" value="send" onclick="CHAT.chat()"/>
  
  <div>receive msg:</div>
  <div id="receiveMsg" style="background-color: gainsboro;"></div>
 </body>
</html>

访问连接：C:\Users\damon\Desktop\netty\WebChat\index.html，我们可以看到效果：

img

接下来，我们需要写 websocket js：

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
ini复制代码<script type="application/javascript">
   
   window.CHAT = {
    socket: null,
    init: function() {
     if (window.WebSocket) {
      CHAT.socket = new WebSocket("ws://192.168.174.145:8088/ws");
      CHAT.socket.onopen = function() {
       console.log("连接建立成功...");
      },
      CHAT.socket.onclose = function() {
       console.log("连接关闭...");
      },
      CHAT.socket.onerror = function() {
       console.log("发生错误...");
      },
      CHAT.socket.onmessage = function(e) {
       console.log("接受到消息：" + e.data);
       var receiveMsg = document.getElementById("receiveMsg");
       var html = receiveMsg.innerHTML;
       receiveMsg.innerHTML = html + "<br/>" + e.data;
      }
     } else {
      alert("浏览器不支持websocket协议...");
     }
    },
    chat: function() {
     var msg = document.getElementById("msgContent");
     CHAT.socket.send(msg.value);
    }
   };
   
   CHAT.init();
   
  </script>

这样，一个简单的 websocket js 就写完了，接下来，我们来演示下。

打开网页，访问 index 页面，我们可以看到连接 websocket 失败，而且会打印发生错误、连接关闭信息，这是因为连接失败时，触发 onerror 事件、onclose 事件：

img

接下来，我们先启动后端 WSServer，同时，刷新页面，可以看到页面显示：连接成功。控制台信息：

img

这里由于我打开了两个页面，所以可以看到后端控制台有打印两次客户端连接的信息，分别对应不同的客户端。接下来，我们输入：Hi,Damon

img

发送后，我们可以看到页面上输出信息：“服务器时间在 2021-05-17T20:05:22.802 接受到消息, 消息为：Hi,Damon”。同时，在另一个客户端窗口，也可以看到输出信息：

img

这是因为后端接收到第一个客户端的请求信息后，将信息转发给所有客户端。接下来，如果我们关闭第一个客户端窗口，则后端会监听到，并且输出：

img

同样，如果我新开一个客户端，并且输入信息，也会被转发到其它客户端：

img

同时，后端控制台会打印对应的请求信息：

img

最后，如果我们主要关闭后端服务，此时，所有的客户端都会失去 socket 连接，会提示：

img

后端整合 Springboot 实现聊天系统

前面介绍了 Websocket 后端处理以及前端的实现逻辑，最后，我们结合 Springboot，来看看后端逻辑的实现。

首先，我们进入依赖 pom：

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
xml复制代码<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.10.RELEASE</version>
        <relativePath/>
    </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</artifactId>
		</dependency>

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

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-configuration-processor</artifactId>
			<optional>true</optional>
		</dependency>
		
		<dependency>
			<groupId>io.netty</groupId>
			<artifactId>netty-all</artifactId>
			<version>4.1.64.Final</version>
		</dependency>
		
		<dependency>
			<groupId>commons-codec</groupId>
			<artifactId>commons-codec</artifactId>
			<version>1.11</version>
		</dependency>
		<dependency>
			<groupId>org.apache.commons</groupId>
			<artifactId>commons-lang3</artifactId>
			<version>3.4</version>
		</dependency>
		<dependency>
			<groupId>org.apache.commons</groupId>
			<artifactId>commons-io</artifactId>
			<version>1.3.2</version>
		</dependency>

		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<version>5.1.41</version>
		</dependency>

		<!-- mybatis -->
		<dependency>
			<groupId>org.mybatis.spring.boot</groupId>
			<artifactId>mybatis-spring-boot-starter</artifactId>
			<version>1.3.1</version>
		</dependency>
		<!--mapper -->
		<dependency>
			<groupId>tk.mybatis</groupId>
			<artifactId>mapper-spring-boot-starter</artifactId>
			<version>1.2.4</version>
		</dependency>
		<!--pagehelper -->
		<dependency>
			<groupId>com.github.pagehelper</groupId>
			<artifactId>pagehelper-spring-boot-starter</artifactId>
			<version>1.2.3</version>
		</dependency>

		<!-- 高性能分布式文件服务器 -->
		<dependency>
		    <groupId>com.github.tobato</groupId>
		    <artifactId>fastdfs-client</artifactId>
		    <version>1.26.2</version>
		</dependency>
		
		<dependency>
		    <groupId>org.springframework</groupId>
		    <artifactId>spring-test</artifactId>
		</dependency>
		
		<!-- 二维码 -->
		<dependency>
		    <groupId>com.google.zxing</groupId>
		    <artifactId>javase</artifactId>
		    <version>3.3.3</version>
		</dependency>

这里主要依赖 Springboot 较高版本 2.3.10.RELEASE，同时，加入了 netty 的依赖，以及数据库 mybatis、fastdfs 等分布式文件服务的依赖。

接下来，我们看看启动类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
less复制代码@SpringBootApplication
// 扫描mybatis mapper包路径
@MapperScan(basePackages="com.damon.mapper")
// 扫描 所有需要的包, 包含一些自用的工具类包 所在的路径
@ComponentScan(basePackages= {"com.damon", "org.n3r.idworker"})
public class Application {
	
	@Bean
	public SpringUtil getSpingUtil() {
		return new SpringUtil();
	}
	
	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}
}

在启动类中，我们看到依据 Springboot 来注入注解，并且，我们扫描注入有些启动 bean。接下来，我们再看看如何引入 Netty 服务端启动：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
typescript复制代码@Component
public class NettyBooter implements ApplicationListener<ContextRefreshedEvent> {

	@Override
	public void onApplicationEvent(ContextRefreshedEvent event) {
		if (event.getApplicationContext().getParent() == null) {
			try {
				WSServer.getInstance().start();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
}

这里主要通过注解@Component 注入一个监听器，同时是在主服务启动的时候来启动 Netty 服务。那么 Netty 的服务实际逻辑在前面也讲过了：

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
csharp复制代码@Component
public class WSServer {

	private static class SingletionWSServer {
		static final WSServer instance = new WSServer();
	}
	
	public static WSServer getInstance() {
		return SingletionWSServer.instance;
	}
	
	private EventLoopGroup mainGroup;
	private EventLoopGroup subGroup;
	private ServerBootstrap server;
	private ChannelFuture future;
	
	public WSServer() {
		mainGroup = new NioEventLoopGroup();
		subGroup = new NioEventLoopGroup();
		server = new ServerBootstrap();
		server.group(mainGroup, subGroup)
			.channel(NioServerSocketChannel.class)
			.childHandler(new WSServerInitialzer());
	}
	
	public void start() {
		this.future = server.bind(8088);
		System.err.println("netty websocket server start over");
	}
}

对于线程组来讲，当客户端与从线程组进行通信后，从线程组会对对应的 Channel 进行处理。同时，每一个 Channel 都是有初始化器，所以这里有 childHandler 函数。channelHandler 的处理器会进行处理 Http、Websocket 等各种协议的请求的支持。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
scala复制代码public class WSServerInitialzer extends ChannelInitializer<SocketChannel> {

	@Override
	protected void initChannel(SocketChannel ch) throws Exception {
		ChannelPipeline pipeline = ch.pipeline();
		
		// websocket 基于http协议，所以要有http编解码器
		pipeline.addLast(new HttpServerCodec());
		// 对写大数据流的支持 
		pipeline.addLast(new ChunkedWriteHandler());
		
		pipeline.addLast(new HttpObjectAggregator(1024*64));
		
		pipeline.addLast(new IdleStateHandler(8, 10, 12));
		// 自定义的空闲状态检测
		pipeline.addLast(new HeartBeatHandler());
    
		pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
		
		// 自定义的handler
		pipeline.addLast(new ChatHandler());
	}

}

到此，所有的后端的技术部分就都讲完了。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%