程序员人生 网站导航

[置顶] 分布式内存网格Hazelcast源码导读

栏目:框架设计时间:2016-06-27 16:27:04

去年项目需要看了hazelcast源码,当时记录的笔记。



Node是节点的抽象,里面包括节点引擎、客户端引擎、分区服务、集群服务、组播服务、连接收理、命令管理、组播属性、节点配置、本地成员、tcp地址、组播地址、连接者、节点初始化器、管理中心、安全上下文、


Config类,包括GroupConfig、NetworkConfig、MapConfig、TopicConfig、QueueConfig、MultiMapConfig、ListConfig、SetConfig、ExecutorConfig、SemaphoreConfig、WanReplicationConfig、ServicesConfig、SecurityConfig、ListenerConfig、PartitionGroupConfig、ManagementCenterConfig、SerializationConfig。

GroupConfig集群用户名及密码DEFAULT_GROUP_PASSWORD = "dev-pass";DEFAULT_GROUP_NAME = "dev";
NetworkConfig网络相干配置,包括InterfacesConfig、JoinConfig、SSLConfig等,其中JoinConfig包括MulticastConfig、TcpIpConfig、AwsConfig。
SecurityConfig客户端登陆相干配置。
WanReplicationConfig集群复制配置,包括WanTargetClusterConfig配置,所有目标节点相干配置。

ClusterServiceImpl用于保护集群各个成员节点,实例化时把本地节点加入成员Map中,MulticastService

Node实例化时根据MulticastConfig使用组播加入1个组,MulticastService提供节点基于组播传递的,使用监听模式每当接收到其他节点传播的消息调用监听器的onMessage,传递的参数为JoinMessage,它由序列化模块提供转化,默许NodeMulticastListener监听器,会对集群的节点校验,是不是是同个集群用户名密码。

MulticastJoiner用于加入集群节点,创建JoinRequest对象用MulticastService发送给其他成员,不断发送加入集群要求直到node里面的masteraddress,如果配置里面指定了targetAddress就不用使用这类不断发送的方式选举主节点。MulticastService负责发送申请加入的组播消息和组播消息接收及处理的工作。

找主节点方式:循环发送JoinRequest消息向组内发送,如果已加入状态且是master的节点接收到后会向外组播JoinMessage,告知其他节点我的组成员信息,还没加入集群的成员则将自己节点的master地址设置为master发出的JoinMessage中的地址。所以JoinMessage只有master才会发出。其他已加入组的成员节点接收到JoinMessage类型消息则直接疏忽,

JoinMessage包括包版本、地址等信息。

SerializationService序列化转化模块,SerializationServiceBuilder生产者,默许字节寄存顺序序列为ByteOrder.BIG_ENDIAN,即便用ByteArrayInputOutputFactory,

Packet为数据包结构,第1字节为版本号,第2字节表示为长度,第3字节表示类型。

SocketConnector会履行连接操作,使用的是阻塞读写。

TcpIpConnectionManager集成ConnectionManager,用于管理TCPIP连接

JoinMessage{
protected byte packetVersion;
    protected int buildNumber;
    protected Address address;
    protected String uuid;
    protected ConfigCheck configCheck;
    protected int memberCount;
}

JoinRequest继承JoinMessage,并添加credentials、tryCount属性。

默许5701作为每一个节点对外暴露的端口。

ReadHandler的handle方法会初始化socketReader,集群成员保护的话使用SocketPacketReader读取报文,

OperationServiceImpl类内部类RemoteOperationProcessor用于ResponseOperation对象
response.beforeRun();
response.run();
response.afterRun();
即会调用JoinRequestOperation的run方法,调用ClusterServiceImpl里面的handleJoinRequest方法,完成成员更新工作,接收获功会向源节点返回SetMasterOperation、MemberInfoUpdateOperation等等更新源节点成员及节点joined状态。

里面1切需要处理的都用run方法,会统1处理。

Node.start--TcpIpConnectionManager.start--InSelectorImpl.run(输入即读取选择器,run不断选择可读的对象并获得attachment,调用它的handle方法,这里的attachment是在TcpIpConnection实例化的时候创建readHandle对象并调用start方法把readHandler当作attachment注册到socketChannel里面的,readHandler调用handle方法,如果是集群协议CLUSTER则实例化SocketPacketReader,SocketPacketReader包括1个PacketReader,默许是DefaultPacketReader,用于将套接字读出来的字节转化成Packet包并调用handlePacket方法处理消息包,如果是集群成员消息包则调用handleMemberPacket方法处理,间接调用NodeEngineImpl的handlePacket方法,有3种不同头部类型:Packet.HEADER_OP、Packet.HEADER_EVENT、Packet.HEADER_WAN_REPLICATION,HEADER_OP表示操作类型,会调用operationService的handleOperation方法处理包消息,OperationThread会1直跑调用OperationRunner处理需要履行的Operation,OperationRunner线程数由hazelcast.operation.thread.count设置,如果⑴则为机器cpu个数,接着会履行operation的beforeRun、run、operation如果设置了响应的话还要履行产生响应内容在返回给调用者、afterRun方法。)--OutSelectorImpl.run(输出即写入选择器)--SocketAcceptor.run--TcpIpConnection.start(TcpIpConnection包括WriteHandler和ReadHandler)--

private void process(Object task) {
        processedCount++;


        if (task instanceof Operation) {
            processOperation((Operation) task);
            return;
        }


        if (task instanceof Packet) {
            processPacket((Packet) task);
            return;
        }


        if (task instanceof PartitionSpecificRunnable) {
            processPartitionSpecificRunnable((PartitionSpecificRunnable) task);
            return;
        }


        throw new IllegalStateException("Unhandled task type for task:" + task);
    }


设计得不错的是它直接支持各种Operation的传递,并履行,本质也是序列化反序列化,然后调用Operation的beforerun、run、afterrun方法,后面还会自动履行handleResponse方法,此方法用于向其他备份节点同步数据,这部份操作是在operation的afterrun之前完成备份。备份工作由OperationBackupHandler完成,backup方法,备份又分为同步备份和异步备份,相加等于总的需要备份数。map操作默许备份1份数据且是同步的,异步的默许为0.异步备份不会阻塞操作。备份的operation是Backup,它的run会履行PutOperation的run方法,即把数据放到缓存中并修改版本,这里的run不会再履行复制操作。
<hazelcast>
  <map name="default">
    <backup-count>1</backup-count>
    <async-backup-count>1</async-backup-count>
  </map>
</hazelcast>
operation的run一样会在备份节点上履行,putoperation其实就是在本地缓存更新值。备份的进程没有1个ack机制,信息传输的可靠性如何保证?

1个节点调用map的put操作时,会在本节点上缓存这个结果,再把operation传输到对应partition的第1个备份节点(这个节点可能就是自己本地节点)上,第1个节点接收到后备份到第2个节点上,所以默许就只有两个备份数据。所以nearcache缓存是可能存在每一个节点上的。

PutOperation的afterRun方法主要是触发1些拦截器,触发各个节点的事件监听器甚么的、更新各个备份节点缓存等等。

IOBalancer平衡io读取写入。

packet报文结构,1byte版本+2byte头部+4byte分区+4byte长度+nbyte消息。

主节点1个1个发给其他成员节点关于成员的消息,其他节点进行更新。

ServiceManagerImpl用于管理所有的远程,启动时会注册经常使用的1些服务,包括以下的服务,甚么map、queue甚么的
    private void registerCoreServices() {
        Node node = nodeEngine.getNode();
        registerService(ClusterServiceImpl.SERVICE_NAME, node.getClusterService());
        registerService(InternalPartitionService.SERVICE_NAME, node.getPartitionService());
        registerService(ProxyServiceImpl.SERVICE_NAME, nodeEngine.getProxyService());
        registerService(TransactionManagerServiceImpl.SERVICE_NAME, nodeEngine.getTransactionManagerService());
        registerService(ClientEngineImpl.SERVICE_NAME, node.clientEngine);
        registerService(QuorumServiceImpl.SERVICE_NAME, nodeEngine.getQuorumService());
    }


    private void registerDefaultServices(ServicesConfig servicesConfig) {
        registerService(MapService.SERVICE_NAME, createService(MapService.class));
        registerService(LockService.SERVICE_NAME, new LockServiceImpl(nodeEngine));
        registerService(QueueService.SERVICE_NAME, new QueueService(nodeEngine));
        registerService(TopicService.SERVICE_NAME, new TopicService());
        registerService(ReliableTopicService.SERVICE_NAME, new ReliableTopicService(nodeEngine));
        registerService(MultiMapService.SERVICE_NAME, new MultiMapService(nodeEngine));
        registerService(ListService.SERVICE_NAME, new ListService(nodeEngine));
        registerService(SetService.SERVICE_NAME, new SetService(nodeEngine));
        registerService(DistributedExecutorService.SERVICE_NAME, new DistributedExecutorService());
        registerService(AtomicLongService.SERVICE_NAME, new AtomicLongService());
        registerService(AtomicReferenceService.SERVICE_NAME, new AtomicReferenceService());
        registerService(CountDownLatchService.SERVICE_NAME, new CountDownLatchService());
        registerService(SemaphoreService.SERVICE_NAME, new SemaphoreService(nodeEngine));
        registerService(IdGeneratorService.SERVICE_NAME, new IdGeneratorService(nodeEngine));
        registerService(MapReduceService.SERVICE_NAME, new MapReduceService(nodeEngine));
        registerService(ReplicatedMapService.SERVICE_NAME, new ReplicatedMapService(nodeEngine));
        registerService(RingbufferService.SERVICE_NAME, new RingbufferService(nodeEngine));
        registerService(XAService.SERVICE_NAME, new XAService(nodeEngine));
        registerCacheServiceIfAvailable();
    }

Map通过AbstractMapServiceFactory创建,使用MapRemoteService处理远程操作,RemoteService服务有两个方法createDistributedObject和destroyDistributedObject方法,终究是通过MapProxyImpl。Map的partition策略在MapContainer里面。

PutOperation用于履行集群节点的put操作。1致性哈希根据key使用MurmurHash哈希计算出结果,再根据分区数(默许271)取余。每一个节点都是使用new ConcurrentHashMap<Data, Record>(1000, 0.75f, 1)寄存记录。

成员组MemberGroup1共有若干个组,多少个成员就多少个组,最大的复制节点数为7,如果成员组小于7则使用成员组数量。

address[271][4]

Address[271][7] 
1、addr0 addr9 addr3 addr4 null null null
2、.
.
.
.
.
271、addr1 addr2 addr3 addr4 null null null


对分配的结果尝试重新分配,把过载的组分配1些成员给不足的组,并检测每一个成员组内的节点数相差不会超过系数1.1,否则重新分组,尽量到达均匀。


终究构成1张表,271个分区每一个分区都对应着若干个复制的成员地址。

int avgPartitionPerGroup = partitionCount / groupSize;
就是说最多4个组,假设5个结点,分组情况为2,1,1,1,则每一个组分到的partition个数为271/4,可能有些组多1个partition。

1共有271条线程处理operation,OperationThread,里面有队列ScheduleQueue,线程会不断处理,ScheduleQueue用于operation履行缓冲队列,里面的有两种队列,normalQueue和priorityQueue,1种用于正常的排队,1种用于设置优先级的队列,take方法会优先从priorityQueue中获得需要优先处理的operation。

有个同步复制、异步复制。

OperationServiceImpl.createInvocationBuilder(此方法有两个,1个用于partition、1个用于指定的address),->  实例化1个InvocationBuilderImpl对象,调用invoke方法会创建Invocation,Invocation包括PartitionInvocation(针对分区)和TargetInvocation(针对指定节点)两种。

在集群中传输1切以Data情势传输。

Map的服务名为hz:impl:mapService。
假设HazelcastInstance履行instance.getMap("customers")则,通过HazelcastInstanceProxy的getMap方法,代理是调HazelcastInstanceImpl的getMap方法,调用getDistributedObject方法,它会通过ProxyService(它代理了所有服务)代理找到MapService,调用mapservice的createDistributedObject方法创建DistributedObject,间接调用MapRemoteService的createDistributedObject方法创建MapProxyImpl,调用MapProxyImpl的put方法把数据放到集群,主要以下操作
final Data key = toData(k, partitionStrategy);
final Data value = toData(v);
final Data result = putInternal(key, value, ttl, timeunit);
return (V) toObject(result);
分别将key和value转化为Data,实际上是序列化,方便网络传输。putInternal方法使用PutOperation,并且要根据key计算出partitionId,接着再完成operation的调用。

SerializationServiceImpl提供各种类型的序列化支持,toData提供由object到Data的转化,toObject提供由Data到object的转化,Data默许是使用DefaultData,DefaultData里面其实就是包括了1个字节数组还有不同的偏移量,例如从几位到几位表示类型,序列化工作其实也跟这类类似,把某1对象的相干信息转化为字节数组,传递到目的地后再根据约定反向组装成指定对象。

OperationThread专门用于履行接收到的要求,process方法,可以有3种要求,包括Operation、Packet、PartitionSpecificRunnable,分别不同的处理逻辑,Operation则直接反序列化后调用beforeRun、run、afterRun等方法。

HealthMonitor是独立1条线程用于监控健康,包括
private class HealthMetrics {
        private final long memoryFree;
        private final long memoryTotal;
        private final long memoryUsed;
        private final long memoryMax;
        private final double memoryUsedOfTotalPercentage;
        private final double memoryUsedOfMaxPercentage;
        //following three load variables are always between 0 and 100.
        private final double processCpuLoad;
        private final double systemLoadAverage;
        private final double systemCpuLoad;
        private final int threadCount;
        private final int peakThreadCount;
        private final long clusterTimeDiff;
        private final int asyncExecutorQueueSize;
        private final int clientExecutorQueueSize;
        private final int queryExecutorQueueSize;
        private final int scheduledExecutorQueueSize;
        private final int systemExecutorQueueSize;
        private final int eventQueueSize;
        private final int pendingInvocationsCount;
        private final double pendingInvocationsPercentage;
        private final int operationServiceOperationExecutorQueueSize;
        private final int operationServiceOperationPriorityExecutorQueueSize;
        private final int operationServiceOperationResponseQueueSize;
        private final int runningOperationsCount;
        private final int remoteOperationsCount;
        private final int proxyCount;
        private final int clientEndpointCount;
        private final int activeConnectionCount;
        private final int currentClientConnectionCount;
        private final int connectionCount;
        private final int ioExecutorQueueSize;
}

PerformanceMonitor表示性能监控,监控的参数包括inSelector的已读取数量,outSelect的已写入事件数量,OperationService相干的性能参数,例如挂起的调用比例、整体调用比例、最大调用数量、271个分区线程已履行数量、271个分区operation线程正在挂起线程(即任务排队队列中的数量),常规operation线程的排队队列数量、常规operation线程已履行数量、响应线程已履行数量、响应线程排队队列数量。

hazelcast核心——Node,包括各种各样重要的基础服务,日志、节点关闭钩子、序列化服务、节点引擎、客户端引擎、分区服务、集群服务、广播服务、连接收理服务、命令服务、配置文件服务、群组属性服务、本节点地址、本地集群成员对象、主节点地址、hazelcast实例援用、日志服务、集群节点加入服务、节点扩大服务、管理中心服务、安全上下文、创建信息服务、版本校订服务、hazelcast线程组。
public class Node {
    private final ILogger logger;
    private final NodeShutdownHookThread shutdownHookThread = new NodeShutdownHookThread("hz.ShutdownThread");
    private final SerializationService serializationService;
    public final NodeEngineImpl nodeEngine;
    public final ClientEngineImpl clientEngine;
    public final InternalPartitionService partitionService;
    public final ClusterServiceImpl clusterService;
    public final MulticastService multicastService;
    public final ConnectionManager connectionManager;
    public final TextCommandServiceImpl textCommandService;
    public final Config config;
    public final GroupProperties groupProperties;
    public final Address address;
    public final MemberImpl localMember;
    private volatile Address masterAddress = null;
    public final HazelcastInstanceImpl hazelcastInstance;
    public final LoggingServiceImpl loggingService;
    private final Joiner joiner;
    private final NodeExtension nodeExtension;
    private ManagementCenterService managementCenterService;
    public final SecurityContext securityContext;
    private final ClassLoader configClassLoader;
    private final BuildInfo buildInfo;
    private final VersionCheck versionCheck = new VersionCheck();
    private final HazelcastThreadGroup hazelcastThreadGroup;
}

NodeEngineImpl作为节点引擎包括了许多服务,重要的例如,事件服务、operation服务、履行服务、等待通知服务、service(内置许多service,例如Map、Queue,用户自定义的服务可配置到hazelcast.xml,启动时会加载进来)管理服务、事务管理服务、代理服务、wan复制服务、包传输服务、证明人服务。
public class NodeEngineImpl implements NodeEngine {


    private final Node node;
    private final ILogger logger;
    private final EventServiceImpl eventService;
    private final OperationServiceImpl operationService;
    private final ExecutionServiceImpl executionService;
    private final WaitNotifyServiceImpl waitNotifyService;
    private final ServiceManagerImpl serviceManager;
    private final TransactionManagerServiceImpl transactionManagerService;
    private final ProxyServiceImpl proxyService;
    private final WanReplicationService wanReplicationService;
    private final PacketTransceiver packetTransceiver;
    private final QuorumServiceImpl quorumService;


}

ServiceManagerImpl用于管理所有服务,启动时默许会实例化核心的service、默许的service,如果用户通过配置文件配置了自定义service则也会实例化。启动时注册行将service实例put到ConcurrentMap中,核心service包括ClusterServiceImpl、InternalPartitionService、ProxyServiceImpl、TransactionManagerServiceImpl、ClientEngineImpl、QuorumServiceImpl。默许service包括MapService、LockService、QueueService、TopicService、ReliableTopicService、MultiMapService、ListService、SetService、DistributedExecutorService、AtomicLongService、AtomicReferenceService、CountDownLatchService、SemaphoreService、IdGeneratorService、MapReduceService、ReplicatedMapService、RingbufferService、XAService。如果允许还将把缓存服务CacheService添加进来。
public final class ServiceManagerImpl implements ServiceManager {


    private final ConcurrentMap<String, ServiceInfo> services = new ConcurrentHashMap<String, ServiceInfo>(20, .75f, 1);


}

节点加入,Joiner负责加入工作,例如广播则使用MulticastJoiner、单播则使用TcpIpJoiner、AWS则使用TcpIpJoinerOverAWS。Node启动时会根据情况启动个线程,
multicast只是做节点发现工作,真实的节点加入工作是交由tcpip做,向主节点发送加入要求,主节点把要求节点添加到成员列表中,然后返回要求节点让它把主节点地址设置为本人。

DefaultSerializers包括DateSerializer、ObjectSerializer、ClassSerializer等等序列化器,实现StreamSerializer的read和write方法完成序列化和反序列化处理。

Date.class, new DateSerializer());
        BigInteger.class, new BigIntegerSerializer());
        BigDecimal.class, new BigDecimalSerializer());
        Externalizable.class, new Externalizer(enableCompression));
        Serializable.class, new ObjectSerializer(enableSharedObject, enableCompression));
        Class.class, new ClassSerializer());
        Enum.class, new EnumSerializer());

DataSerializable.class, dataSerializerAdapter);
        Portable.class, portableSerializerAdapter);
        Byte.class, new ByteSerializer());
        Boolean.class, new BooleanSerializer());
        Character.class, new CharSerializer());
        Short.class, new ShortSerializer());
        Integer.class, new IntegerSerializer());
        Long.class, new LongSerializer());
        Float.class, new FloatSerializer());
        Double.class, new DoubleSerializer());
        byte[].class, new TheByteArraySerializer());
        char[].class, new CharArraySerializer());
        short[].class, new ShortArraySerializer());
        int[].class, new IntegerArraySerializer());
        long[].class, new LongArraySerializer());
        float[].class, new FloatArraySerializer());
        double[].class, new DoubleArraySerializer());
        String.class, new StringSerializer());

每一个service都有自己的context,例如mapservice的MapServiceContext,它里面是保存了1份partition映照表的副本,在底层完成迁移之前其实不会更新,固然底层的map数据也不会1边迁移1边删除,而是复制1份进行删除



喜欢java的同学可以交个朋友:





------分隔线----------------------------
------分隔线----------------------------

最新技术推荐