Zeebe Gateway源码之StandaloneGateway创建AtomixCluster详解(一)

// ... ...
// StandaloneGateway就是程序的入口
exec "$JAVACMD" $JAVA_OPTS -Xms128m -XX:+ExitOnOutOfMemoryError -Dfile.encoding=UTF-8 \
  -classpath "$CLASSPATH" \
  -Dapp.name="gateway" \
  -Dapp.pid="$$" \
  -Dapp.repo="$REPO" \
  -Dapp.home="$BASEDIR" \
  -Dbasedir="$BASEDIR" \
  io.camunda.zeebe.gateway.StandaloneGateway 

// ... ...  

// **********************************************************
// 配置了sprig自动扫描如下package,咱先不管它,先看main函数
// **********************************************************
@SpringBootApplication(
    scanBasePackages = {
      "io.camunda.zeebe.gateway",
      "io.camunda.zeebe.shared",
      "io.camunda.zeebe.util.liveness"
    })
// **********************************************************
// 配置了ConfigurationPropertie的扫描
// **********************************************************
@ConfigurationPropertiesScan(basePackages = {"io.camunda.zeebe.gateway", "io.camunda.zeebe.shared"})
public class StandaloneGateway
    //  **********************************************************
	// 要稍微注意下,实现了哪些接口,其中:CommandLineRunner是Spring容器启动完成之后的一个Callback函数.
	//  **********************************************************
    implements CommandLineRunner, ApplicationListener<ContextClosedEvent>, CloseableSilently {
	// ... ... 
}

public static void main(final String[] args) {
	// 1. 为线程配置当出现异常时的处理
	Thread.setDefaultUncaughtExceptionHandler(FatalErrorHandler.uncaughtExceptionHandler(Loggers.GATEWAY_LOGGER));

	// 2. 配置banner
	System.setProperty("spring.banner.location", "classpath:/assets/zeebe_gateway_banner.txt");
	// 3. 配置容器为servlet/profiles
	final var application = new SpringApplicationBuilder(StandaloneGateway.class)
			.web(WebApplicationType.SERVLET)
			.logStartupInfo(true)
			.profiles(Profile.GATEWAY.getId())
			.build(args);
    // 4. 运行容器
	application.run();
}

@Autowired
public StandaloneGateway(
   // 依赖注入:GatewayCfg/SpringGatewayBridge/ActorClockConfiguration
   // ActorClockConfiguration是一个时钟管理,我们不分析,直接扔掉不管理.
  final GatewayCfg configuration,
  final SpringGatewayBridge springGatewayBridge,
  final ActorClockConfiguration clockConfig) {
	this.configuration = configuration;
	this.springGatewayBridge = springGatewayBridge;
	this.clockConfig = clockConfig;
}

@Component
@ConfigurationProperties(prefix = "zeebe.gateway")
public class GatewayCfg {

  private NetworkCfg network = new NetworkCfg();
  private ClusterCfg cluster = new ClusterCfg();
  private ThreadsCfg threads = new ThreadsCfg();
  private SecurityCfg security = new SecurityCfg();
  private LongPollingCfg longPolling = new LongPollingCfg();
  private List<InterceptorCfg> interceptors = new ArrayList<>();
  private boolean initialized = false;
  // ... ... 
}

@Component
public class SpringGatewayBridge {

  private Supplier<Status> gatewayStatusSupplier;
  private Supplier<Optional<BrokerClusterState>> clusterStateSupplier;
  private Supplier<BrokerClient> brokerClientSupplier;
  
  // ... ... 
  
  // ************************************************
  // 注册:BrokerClient
  // ************************************************
  public void registerBrokerClientSupplier(final Supplier<BrokerClient> brokerClientSupplier) {
    this.brokerClientSupplier = brokerClientSupplier;
  }
  
  // ************************************************
  // 获取:BrokerClient
  // ************************************************
  public Optional<BrokerClient> getBrokerClient() {
    return Optional.ofNullable(brokerClientSupplier).map(Supplier::get);
  }
}

// 通过Spring容器注入的GatewayCfg,主要用于对配置文件的管理.
private final GatewayCfg configuration;

private ActorScheduler actorScheduler;


public void run(final String... args) throws Exception {
    // ... ...
    // ****************************************************************
	// 看到了我们重点:AtomixCluster
	// Atomix是RAFT和Gossip的实现,咱先不去详细分析RAFT和Gossip.
	// ****************************************************************
	atomixCluster = createAtomixCluster(configuration.getCluster());
	// ... ...
	atomixCluster.start();
}

// ********************************************************
private AtomixCluster createAtomixCluster(final ClusterCfg config) {
	// ********************************************************
	// 创建成员协议,这个协议实际是一个Gossip协议
	// createMembershipProtocol函数的内容,请参考下面(10)小点,大部份参数是默认的,只是配置了一个:BroadcastDispute
	// ********************************************************
	final var membershipProtocol = createMembershipProtocol(config.getMembership());
	final var builder =
		AtomixCluster.builder()
		     // gateway
			.withMemberId(config.getMemberId())
			// gateway:26502
			.withAddress(Address.from(config.getHost(), config.getPort()))
			// zeebe-cluster
			.withClusterId(config.getClusterName())
			// ****************************************************************
			// 配置node节点(broker)的成员列表
			// ****************************************************************
			.withMembershipProvider(
				BootstrapDiscoveryProvider.builder()
				     //  node-1:26502
					.withNodes(Address.from(config.getContactPoint()))
					.build())
			// 配置上面的成员协议信息		
			.withMembershipProtocol(membershipProtocol)
			// 配置消息压缩信息
			.withMessageCompression(config.getMessageCompression());

     // 判断Security是否开启
	if (config.getSecurity().isEnabled()) {
	  applyClusterSecurityConfig(config, builder);
	}
    
	// 构建AtomixCluster
	return builder.build();
}

private GroupMembershipProtocol createMembershipProtocol(final MembershipCfg config) {
	
	return SwimMembershipProtocol.builder()
	     // PT10S
		.withFailureTimeout(config.getFailureTimeout())
		// PT0.25S
		.withGossipInterval(config.getGossipInterval())
		// PT1S
		.withProbeInterval(config.getProbeInterval())
		// PT0.1S
		.withProbeTimeout(config.getProbeTimeout())
		// true
		.withBroadcastDisputes(config.isBroadcastDisputes())
		// false
		.withBroadcastUpdates(config.isBroadcastUpdates())
		// 2
		.withGossipFanout(config.getGossipFanout())
		// false
		.withNotifySuspect(config.isNotifySuspect())
		// 3
		.withSuspectProbes(config.getSuspectProbes())
		// PT10S
		.withSyncInterval(config.getSyncInterval())
		.build();
}