[Bug][Refactor][issue-3157]use cas to avoid thread safe problem (#3158)

Co-authored-by: lgcareer <18610854716@163.com>

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingServer.java

@@ -119,44 +119,39 @@ public class NettyRemotingServer {
      *  server start
      */
     public void start(){
-
-        if(this.isStarted.get()){
-            return;
-        }
-
-        this.serverBootstrap
-                .group(this.bossGroup, this.workGroup)
-                .channel(NioServerSocketChannel.class)
-                .option(ChannelOption.SO_REUSEADDR, true)
-                .option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog())
-                .childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive())
-                .childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay())
-                .childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize())
-                .childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize())
-                .childHandler(new ChannelInitializer<NioSocketChannel>() {
-
-                    @Override
-                    protected void initChannel(NioSocketChannel ch) throws Exception {
-                        initNettyChannel(ch);
-                    }
-                });
-
-        ChannelFuture future;
-        try {
-            future = serverBootstrap.bind(serverConfig.getListenPort()).sync();
-        } catch (Exception e) {
-            logger.error("NettyRemotingServer bind fail {}, exit",e.getMessage(), e);
-            throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
-        }
-        if (future.isSuccess()) {
-            logger.info("NettyRemotingServer bind success at port : {}", serverConfig.getListenPort());
-        } else if (future.cause() != null) {
-            throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()), future.cause());
-        } else {
-            throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
+        if (isStarted.compareAndSet(false, true)) {
+            this.serverBootstrap
+                    .group(this.bossGroup, this.workGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .option(ChannelOption.SO_REUSEADDR, true)
+                    .option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog())
+                    .childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive())
+                    .childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay())
+                    .childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize())
+                    .childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize())
+                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
+
+                        @Override
+                        protected void initChannel(NioSocketChannel ch) throws Exception {
+                            initNettyChannel(ch);
+                        }
+                    });
+
+            ChannelFuture future;
+            try {
+                future = serverBootstrap.bind(serverConfig.getListenPort()).sync();
+            } catch (Exception e) {
+                logger.error("NettyRemotingServer bind fail {}, exit",e.getMessage(), e);
+                throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
+            }
+            if (future.isSuccess()) {
+                logger.info("NettyRemotingServer bind success at port : {}", serverConfig.getListenPort());
+            } else if (future.cause() != null) {
+                throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()), future.cause());
+            } else {
+                throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
+            }
         }
-        //
-        isStarted.compareAndSet(false, true);
     }
 
     /**