[TiO websocket] IV. the TiO websocket server implements the custom cluster mode

t-io The cluster function has been implemented in , be based on Redis A cluster implemented in the publish subscribe mode of . Of course , We can also customize the cluster mode , As long as it is a component that can implement publish and subscribe , Almost all , for example ：rabbitmq,activemq etc. . This time we use the simple activemq

1、 introduce maven rely on

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

Introduced activemq Of information

2、 Turn on ActiveMQ

Annotate the project startup class @EnableJms

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.jms.annotation.EnableJms;

/** *  Turn on activemq */
@EnableJms
@SpringBootApplication
public class TioApplication {
    

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

}

3、ActiveMQ Configuration information

spring:
  jms:
    pub-sub-domain: true
  activemq:
    broker-url: tcp://127.0.0.1:61616
    #  user 、 password 
    user: admin
    password: admin
    packages:
      trust-all: true

packages.trust-all Must be configured to true, Otherwise, it will report a mistake

4、ActiveMQ Configuration class

import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.jms.Topic;

@Configuration
public class ActiveMqConfig {
    

    /** *  The theme  */
    public static final String TOPICNAME = "tio_ws_spring_boot_starter";

    @Bean
    public Topic topic() {
    
        return new ActiveMQTopic(TOPICNAME);
    }
}

We injected ActiveMQ A theme of , Later, we will implement the cluster function based on this topic

5、 Cluster publish subscription

import com.asurplus.tio.common.activemq.ActiveMqConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Component;
import org.tio.cluster.TioClusterMessageListener;
import org.tio.cluster.TioClusterTopic;
import org.tio.cluster.TioClusterVo;

import javax.jms.Topic;

/** *  Publish and subscribe implementation  */
@Component
public class MyTioClusterTopic implements TioClusterTopic {
    

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    @Autowired
    private Topic topic;

    //  Storage tioClusterMessageListener Instance information , The consumption message needs to call its onMessage() Interface 
    private TioClusterMessageListener tioClusterMessageListener;

    public void addMessageListener(TioClusterMessageListener tioClusterMessageListener) {
    
        this.tioClusterMessageListener = tioClusterMessageListener;
    }

    /** *  Cluster messages are sent from here to mq * * @param tioClusterVo */
    @Override
    public void publish(TioClusterVo tioClusterVo) {
    
        jmsMessagingTemplate.convertAndSend(topic, tioClusterVo);
    }

    /** *  Listen to the message  * * @param tioClusterVo */
    @JmsListener(destination = ActiveMqConfig.TOPICNAME)
    public void receive(TioClusterVo tioClusterVo) {
    
        if (tioClusterMessageListener != null) {
    
            //  Send the consumption message to onMessage In this way 
            this.tioClusterMessageListener.onMessage(ActiveMqConfig.TOPICNAME, tioClusterVo);
        }
    }
}

The core is to set up a listener , When you get a message , Publish to activemq, After being heard by the supervisor, specific business logic will be carried out

6、 Cluster configuration

import com.asurplus.tio.websocket.handle.MyWsMsgHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.tio.cluster.TioClusterConfig;
import org.tio.server.ServerTioConfig;
import org.tio.websocket.server.WsServerStarter;

import java.io.IOException;

@Configuration
public class WebSocketConfig {
    

    @Autowired
    private MyTioClusterTopic myTioClusterTopic;
    @Autowired
    private MyWsMsgHandler myWsMsgHandler;

    /** * TIO-WEBSOCKET  Configuration information  */
    public static ServerTioConfig serverTioConfig;

    /** *  Cluster configuration  * * @return */
    @Bean
    public TioClusterConfig tioClusterConfig() {
    
        //  Set up Tio Cluster configuration , Set the cluster message processor 
        TioClusterConfig clusterConfig = new TioClusterConfig(myTioClusterTopic);
        //  Whether all connections are clustered （ The same ip Whether it will be distributed on different machines ）,false: No cluster , Default cluster 
        clusterConfig.setCluster4all(true);
        // bsid Cluster or not （ stay A Whether the client on the machine can pass bsid Send a message to B The client on the machine ）,false: No cluster , Default cluster 
        clusterConfig.setCluster4bsId(true);
        // id Cluster or not （ stay A Whether the client on the machine can pass channelId Send a message to B The client on the machine ）,false: No cluster , Default cluster 
        clusterConfig.setCluster4channelId(true);
        //  Whether the group is clustered （ Whether the same group will be distributed on different machines ）,false: No cluster , Cluster is not selected by default 
        clusterConfig.setCluster4group(true);
        // ip Cluster or not （ The same ip Whether it will be distributed on different machines ）,false: No cluster , Default cluster 
        clusterConfig.setCluster4ip(true);
        //  Whether the user is clustered （ Whether the same user will be distributed on different machines ）,false: No cluster , Default cluster 
        clusterConfig.setCluster4user(true);
        return clusterConfig;
    }

    /** *  Start class configuration  * * @return * @throws IOException */
    @Bean
    public WsServerStarter wsServerStarter() throws IOException {
    
        //  Set up the processor 
        WsServerStarter wsServerStarter = new WsServerStarter(6789, myWsMsgHandler);
        //  Get ServerTioConfig
        serverTioConfig = wsServerStarter.getServerTioConfig();
        //  Set timeout 
        serverTioConfig.setHeartbeatTimeout(600000);
        //  Set the cluster configuration file 
        serverTioConfig.setTioClusterConfig(tioClusterConfig());
        //  start-up 
        wsServerStarter.start();
        return wsServerStarter;
    }
}

t-io Whether the cluster of is enabled , Depending on TioClusterConfig Whether to configure , Now our cluster function is configured

7、 Package test

We need to pack two bags for testing , Use two different id Log in to two different websocket The server , See if it can send and receive messages normally

Testing process , A little …

It is feasible to test in person

If you find deficiencies in reading , Welcome to leave a message ！！！