一个服务存在多个实例时，Netty连接通过网关会被负载均衡连接到其中任意一个实例上。

而当一个服务实例发送消息时，连接另一个实例的客户端就会收不到消息

为了解决这个问题，该库提供了一种解决方案，开箱即用

只需要添加一个配置注解，就可以像单体应用一样使用Netty

也可以通过自定义来支持更复杂的业务

implementation 'com.github.linyuzai:concept-netty-loadbalance-spring-boot-starter:${version}'

<dependency>
  <groupId>com.github.linyuzai</groupId>
  <artifactId>concept-netty-loadbalance-spring-boot-starter</artifactId>
  <version>${version}</version>
</dependency>

在启动类上添加注解@EnableNettyLoadBalanceConcept启用功能

@EnableNettyLoadBalanceConcept
@SpringBootApplication
public class NettyServiceApplication {

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

在Netty中配置NettyLoadBalanceHandler来接管连接

@Component
public class NettySampleServer {

    @Autowired
    private NettyLoadBalanceConcept concept;

    public void start(int port) {
        EventLoopGroup boss = new NioEventLoopGroup(1);
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            pipeline.addLast(new LineBasedFrameDecoder(1024));
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new StringDecoder());
                            //将连接交由 NettyLoadBalanceHandler 管理
                            pipeline.addLast(new NettyLoadBalanceHandler(concept));
                        }
                    });
            ChannelFuture future = bootstrap.bind(port).sync();
            future.channel().closeFuture().sync();
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

注入NettyLoadBalanceConcept即可跨实例发送消息

@RestController
@RequestMapping("/netty")
public class NettyController {

    @Autowired
    private NettyLoadBalanceConcept concept;

    @RequestMapping("/send")
    public void send(@RequestParam String msg) {
        concept.send(msg);
    }
}

concept:
  netty:
    server: #服务配置
      message:
        retry:
          times: 0 #客户端重试次数，默认不重试
          period: 0 #客户端重试间隔，单位ms，默认0ms
    load-balance: #负载均衡（转发）配置
      subscriber-master: none #主订阅器，默认无
      subscriber-slave: none #从订阅器，默认无
      message:
        retry:
          times: 0 #转发重试次数，默认不重试
          period: 0 #转发重试间隔，单位ms，默认0ms
      heartbeat: #心跳配置
        enabled: true #是否启用心跳，默认true
        period: 60000 #心跳间隔，单位ms，默认1分钟
        timeout: 210000 #超时时间，单位ms，默认3.5分钟，3次心跳间隔
    executor:
      thread-pool-size: 1 #线程池大小，默认1

通过Redis和MQ等中间件转发消息

由于本库支持多种连接（当前包括WebSocket和Netty和SSE）同时配置，所以引入连接域来进行限制。

在自定义组件（所有组件）时需要指定该组件所适配的连接类型（NettyScoped.NAME/WebSocketScoped.NAME/SseScoped.NAME）

可通过重写boolean support(String scope)方法或是调用addScopes(String... scopes)来配置

可以实现NettyEventListener来监听事件

可以实现NettyLifecycleListener来监听生命周期（连接发布/连接关闭）

可以实现NettyMessageHandler来处理消息

可以继承NettyMessageCodecAdapter来添加消息编解码器

可以配置2种订阅转发方式提高容错

如以Kafka为主，Redis为从

当Kafka转发消息失败后，会切换到Redis重新转发，并开始对Kafka进行定时的的心跳检测

等到Kafka心跳检测正常，则重新切回到Kafka

抛出MessageTransportException将会触发切换

concept:
  netty:
    load-balance:
      subscriber-master: none #主订阅者器，默认无
      subscriber-slave: none #从订阅器，默认无

通过Kafka/RabbitMQ转发消息时可能会重复消费

提供MessageIdempotentVerifier抽象messageId生成以及验证是否重复

默认缓存所有messageId在内存中（存在缓存越来越大的问题，建议自定义使用Redis或数据库存储）

可通过MessageIdempotentVerifierFactory自定义并注入Spring生效

可通过自定义ConnectionSelector来实现消息的准确发送

如果在复用EventLoopGroup的情况下，可以配置分组方便管理

在添加NettyLoadBalanceHandler时传入分组

pipeline.addLast(new NettyLoadBalanceHandler(concept, "group1"));

使用GroupMessage给某个分组发送消息

@RestController
@RequestMapping("/netty")
public class NettyController {

    @Autowired
    private NettyLoadBalanceConcept concept;

    @RequestMapping("/send-group")
    public void sendGroup(@RequestParam String msg) {
        concept.send(new GroupMessage(msg, "group1"));
    }
}

默认情况下，ConnectionSelector对于每次发送消息都只会有一个生效（只能生效一种过滤条件）

对于ConnectionSelector提供扩展FilterConnectionSelector

将会作为一个过滤器来支持多种条件，即组合条件模式

为解决同一个数据多次编码成相同的数据

提供ReusableMessage来缓存编码后的数据

ObjectMessage message = new ObjectMessage("消息数据");
ReusableMessage reusaeble = message.toReusableMessage();
concept.send(reusaeble);

默认情况下按顺序循环发送

可以注入CompletableFutureMessageSenderFactory实现并发发送

或是自定义实现(Abstract)MessageSenderFactory

实现NettyMessageHandler来处理客户端发送的消息

默认配置的编解码器都是转JSON

可以通过NettyMessageCodecAdapter或(Abstract)MessageCodecAdapter自定义

所有组件均可自定义扩展（可能需要指定Order来保证自定义的组件生效）

ConnectionRepository用于缓存连接实例

默认使用Map<String, Map<Object, Connection>> connections = new ConcurrentHashMap<>();缓存在内存中

可自定义ConnectionRepositoryFactory注入容器生效

ConnectionServerManager用于获取其他服务实例信息（ws双向连接中使用）和自身服务信息

默认使用DiscoveryClient和Registration来获得信息

可自定义ConnectionServerManagerFactory注入容器生效

ConnectionSubscriber用于订阅其他服务的消息

提供配置文件配置

可自定义(Abstract)ConnectionSubscriberFactory或(Abstract)MasterSlaveConnectionSubscriberFactory注入容器生效

ConnectionFactory用于扩展Connection（如WebSocketConnection/NettyConnection/SseConnection）

可自定义ConnectionFactory注入容器生效

ConnectionSelector用于在发送消息时选择发送给哪些连接

可自定义ConnectionSelector或FilterConnectionSelector注入容器生效

MessageFactory用于适配创建消息

可自定义MessageFactory注入容器生效

MessageCodecAdapter用于适配各种连接类型的消息编码器和消息解码器

可自定义(Abstract)MessageCodecAdapter注入容器生效

MessageRetryStrategyAdapter用于适配各种连接类型的消息重试策略

消息重试不对ping和pong生效

可自定义(Abstract)MessageRetryStrategyAdapter注入容器生效

MessageIdempotentVerifier用于生成消息ID以及校验消息是否处理

可自定义MessageIdempotentVerifierFactory注入容器生效

ScheduledExecutor用于执行各种延时/定时任务，如心跳等

可自定义ScheduledExecutorFactory注入容器生效

ConnectionLogger用于打印日志

默认使用Spring的日志库

可自定义ConnectionLoggerFactory注入容器生效

ConnectionEventPublisher用于发布事件

默认支持@EventListener

可自定义ConnectionEventPublisherFactory注入容器生效

NettyEventListener用于监听事件

使用WebSocketNettyLoadBalanceHandler代替NettyLoadBalanceHandler

• SpringBoot3 RabbitEndpoint 兼容问题

• 发送对象消息转发变成空字符串的问题

• 支持转发反序列化
• 内存幂等校验添加固定时间清理功能
• 新增 MessageSender 可自定义消息发送策略

• 可复用消息（原PooledMessage）

• 兼容 Spring3 Redis