Raft算法实现复制

红太狼 2023-06-19 02:19 32阅读 0赞

**一、CopyCat技术与Atomix技术的区别**  
`Atomix`是一种以多种不同方式解决常见分布式系统问题的工具。它没有解决问题的观点，而是提供了解决问题的原始方法。它提供的原语的一些示例是：分布式数据结构（地图，集合，树，计数器，值等）、分布式通信（直接，发布-订阅等）、分布式协调（锁定，领导人选举，信号灯，障碍等）之类的解决分布式工具。`Atomix`可以对一个事件驱动的框架，用于使用多种成熟的分布式系统协议来协调容错分布式系统。它提供了构建模块，可以解决许多常见的分布式系统问题，包括集群管理，异步消息传递，组成员身份，组长选举，分布式并发控制，分区和复制。  
`CopyCat`可以说，Atomix中一块技术的实现部分，也是Atomix的老版本。`Copycat 2.X`版本是现在是`atomix-raft`， 并且包括对Copycat 1.x的多种改进：每个集群有多个状态机、每个客户多个会话、无索引的内存映射日志、每个状态机的快照、框架不可、序列化、分区等。  
**二、Raft算法复制**  
`Raft`是由`Atomix`实施以在节点之间共享状态的特定分布式系统协议`。Atomix`核心依赖几种不同类型的协议来进行状态复制，范围从强一致性到弱一致性。  
该`Raft`协议是`atomix`用于强一致的，分区容错元达成共识的协议。`Atomix`为基于共识的基元提供了`Raft`协议的成熟的自定义实现。`Raft`协议的核心是管理对原语更改的持久复制日志。这是通过选举领导者并将更改同步复制到关注者来完成的。通过仅选举具有所有最新更改的领导者来保持一致性。但是，`Raft`协议的一个重要属性是，只有在大多数群集可用时，它才能取得进展。如果是网络分区，则只有分区的多数方会继续取得进展。

**三、客户端复制服务**

package com.citydo.sentinel.atomix;
    
    import com.citydo.sentinel.copycat.GetQuery;
    import com.citydo.sentinel.copycat.PutCommand;
    import io.atomix.catalyst.transport.Address;
    import io.atomix.catalyst.transport.netty.NettyTransport;
    import io.atomix.copycat.client.CopycatClient;
    import java.util.Arrays;
    import java.util.Collection;
    import java.util.concurrent.CompletableFuture;
    
    public class AtomixClient { 
        public static void main(String[] args){ 
    
            CopycatClient.Builder builder = CopycatClient.builder();
    
            builder.withTransport(NettyTransport.builder()
                    .withThreads(2)
                    .build());
    
            CopycatClient client = builder.build();
    
            //客户端注册命令
            client.serializer().register(PutCommand.class);
            client.serializer().register(GetQuery.class);
    
    
            //集群的ip以及端口
            Collection<Address> cluster = Arrays.asList(
                    new Address("127.0.0.1", 5000),
                    new Address("127.0.0.1", 5001),
                    new Address("127.0.0.1", 5002)
            );
    
            CompletableFuture<CopycatClient> future = client.connect(cluster);
            future.join();
    
            //使用PutCommand提交三个键值对
            CompletableFuture[] futures = new CompletableFuture[3];
            futures[0] = client.submit(new PutCommand("one", "Hello world!"));
            futures[1] = client.submit(new PutCommand("two", "Hello world!"));
            futures[2] = client.submit(new PutCommand("three", "Hello world!"));
    
            //等待集群完成一致性的复制后，打印完成的结果
            CompletableFuture.allOf(futures).thenRun(() -> System.out.println("Commands completed!"));
    
            //客户端提交查询
            client.submit(new GetQuery("one")).thenAccept(result -> { 
                System.out.println("one is: " + result);
            });
        }
    }

**四、服务端注册**

package com.citydo.sentinel.atomix;
    
    import com.citydo.sentinel.copycat.GetQuery;
    import com.citydo.sentinel.copycat.MapstateMachine;
    import com.citydo.sentinel.copycat.PutCommand;
    import io.atomix.catalyst.transport.Address;
    import io.atomix.catalyst.transport.netty.NettyTransport;
    import io.atomix.copycat.server.CopycatServer;
    import io.atomix.copycat.server.storage.Storage;
    import io.atomix.copycat.server.storage.StorageLevel;
    
    import java.io.File;
    import java.util.concurrent.CompletableFuture;
    
    public class AtomixOne { 
    
        public static void main(String[] args){ 
    
            //设置server_1的地址和端口
            Address address = new Address("127.0.0.1", 8001);
            CopycatServer server = CopycatServer.builder(address)
                    .withStateMachine(MapstateMachine::new)
                    .withTransport(NettyTransport.builder()
                            .withThreads(4)
                            .build())
                    .withStorage(Storage.builder()
                            .withDirectory(new File("logs"))
                            .withStorageLevel(StorageLevel.DISK)
                            .build())
                    .build();
    
            server.serializer().register(PutCommand.class);
            server.serializer().register(GetQuery.class);
    
            //启动服务器
            CompletableFuture<CopycatServer> future = server.bootstrap();
            future.join();
    
        }
    }

package com.citydo.sentinel.atomix;
    
    import com.citydo.sentinel.copycat.GetQuery;
    import com.citydo.sentinel.copycat.MapstateMachine;
    import com.citydo.sentinel.copycat.PutCommand;
    import io.atomix.catalyst.transport.Address;
    import io.atomix.catalyst.transport.netty.NettyTransport;
    import io.atomix.copycat.server.CopycatServer;
    import io.atomix.copycat.server.storage.Storage;
    import io.atomix.copycat.server.storage.StorageLevel;
    
    import java.io.File;
    import java.util.concurrent.CompletableFuture;
    
    public class AtomixThree { 
        public static void main(String[] args){ 
    
            //设置server_1的地址和端口
            Address address = new Address("127.0.0.1", 8003);
            CopycatServer server = CopycatServer.builder(address)
                    .withStateMachine(MapstateMachine::new)
                    .withTransport(NettyTransport.builder()
                            .withThreads(4)
                            .build())
                    .withStorage(Storage.builder()
                            .withDirectory(new File("logs"))
                            .withStorageLevel(StorageLevel.DISK)
                            .build())
                    .build();
    
            server.serializer().register(PutCommand.class);
            server.serializer().register(GetQuery.class);
    
            //启动服务器
            CompletableFuture<CopycatServer> future = server.bootstrap();
            future.join();
    
        }
    }

package com.citydo.sentinel.atomix;
    
    import com.citydo.sentinel.copycat.GetQuery;
    import com.citydo.sentinel.copycat.MapstateMachine;
    import com.citydo.sentinel.copycat.PutCommand;
    import io.atomix.catalyst.transport.Address;
    import io.atomix.catalyst.transport.netty.NettyTransport;
    import io.atomix.copycat.server.CopycatServer;
    import io.atomix.copycat.server.storage.Storage;
    import io.atomix.copycat.server.storage.StorageLevel;
    
    import java.io.File;
    import java.util.concurrent.CompletableFuture;
    
    public class AtomixTwo { 
        public static void main(String[] args){ 
    
            //设置server_1的地址和端口
            Address address = new Address("127.0.0.1", 8002);
            CopycatServer server = CopycatServer.builder(address)
                    .withStateMachine(MapstateMachine::new)
                    .withTransport(NettyTransport.builder()
                            .withThreads(4)
                            .build())
                    .withStorage(Storage.builder()
                            .withDirectory(new File("logs"))
                            .withStorageLevel(StorageLevel.DISK)
                            .build())
                    .build();
    
            server.serializer().register(PutCommand.class);
            server.serializer().register(GetQuery.class);
    
            //启动服务器
            CompletableFuture<CopycatServer> future = server.bootstrap();
            future.join();
    
        }
    }

package com.citydo.sentinel.atomix;
    
    import io.atomix.cluster.MemberId;
    import io.atomix.core.Atomix;
    import java.util.concurrent.CompletableFuture;
    
    /** * @author nick */
    public class AtomixUtils { 
    
    
         private void atomix(){ 
              //引导集群
              Atomix atomix = Atomix.builder()
                      .withMemberId("member1")
                      .withAddress("10.192.19.181:5679")
                      .withMulticastEnabled()
                      .build();
              atomix.start().join();
    
              //订阅消息
              atomix.getCommunicationService().subscribe("test", message -> { 
                   return CompletableFuture.completedFuture(message);
              });
    
              //发送消息
              atomix.getCommunicationService().send("test", "Hello world!", MemberId.from("foo")).thenAccept(response -> { 
                   System.out.println("Received " + response);
              });
    
         }
    }