标签：cassandra   spark   大数据   

cassandra 服务启动流程

1.  setup

1)   CassandraDaemon ->main

 publicstaticvoidmain(String[]args)

   {

        instance.activate();

   }

2)   系统参数初始化

    配置文件的读取和解析都是在org.apache.cassandra.config.DatabaseDescriptor 类中完成的，这个类的作用非常简单，就是读取配置文件中各个配置项所定义的值，经过简单的验证，符合条件就将其值赋给 DatabaseDescriptor 的私有静态常量。具体的实现过程如下：

DatabaseDescriptor.hasLargeAddressSpace() //加载了系统设置，静态变量，加载了系统的默认参数：

->applyConfig(loadConfig());

下面是获取系统需要的表格，

->// Hardcoded system keyspaces

  List<KSMetaData> systemKeyspaces =Arrays.asList(KSMetaData.systemKeyspace());

       assert systemKeyspaces.size() == Schema.systemKeyspaceNames.size();

       for (KSMetaData ksmd : systemKeyspaces)

            Schema.instance.load(ksmd);

    ->每一个表格写入到Schema

       /**

     *Load specific keyspace into Schema

     *

     *@param keyspaceDef The keyspace to load up

     *

     *@return self to support chaining calls

    */

   public Schema load(KSMetaData keyspaceDef)

    {

       for (CFMetaData cfm : keyspaceDef.cfMetaData().values())

           load(cfm);

       setKeyspaceDefinition(keyspaceDef);

       return this;

    }

最后的cfm都是存放到：private final ConcurrentBiMap<Pair<String,String>, UUID> cfIdMap = new ConcurrentBiMap<>();

上面的这段代码获取到系统需要默认的表格，但是这边还没有创建表格。

3)  keyspacemeta

a)        for(MemoryPoolMXBean pool:ManagementFactory.getMemoryPoolMXBeans())

logger.info("{} {}: {}",pool.getName(), pool.getType(), pool.getPeakUsage());//输出cassandra jvm的所有pool的信息

输出结果如下：

INFO 07:20:32 Heap size: 124780544/954728448

INFO 07:21:12 Code Cache Non-heap memory: init = 2555904(2496K) used =828800(809K) committed = 2555904(2496K) max = 50331648(49152K)

INFO 07:21:28 PS Eden Space Heap memory: init = 33030144(32256K) used =33030144(32256K) committed = 33030144(32256K) max = 347602944(339456K)

INFO 07:21:29 PS Survivor Space Heap memory: init = 5242880(5120K) used =5227632(5105K) committed = 5242880(5120K) max = 5242880(5120K)

INFO 07:22:43 PS Old Gen Heap memory: init = 86507520(84480K) used =351840(343K) committed = 86507520(84480K) max = 715653120(698880K)

INFO 07:22:49 PS Perm Gen Non-heap memory: init = 22020096(21504K) used =16674864(16284K) committed = 22020096(21504K) max = 85983232(83968K)

b)        检查目录是否存在和权限

c)        启动内存初始化

private CacheService()

 {

............................................

       keyCache = initKeyCache();

       //keycache初始化

       rowCache = initRowCache();

        // rowCache初始化

       counterCache = initCounterCache();

      // counterCache处理化

   }

下面我们分析keycache处理化的实现过程：

private AutoSavingCache<KeyCacheKey,RowIndexEntry> initKeyCache() {

  longkeyCacheInMemoryCapacity = DatabaseDescriptor.getKeyCacheSizeInMB() * 1024 *1024;

  ICache<KeyCacheKey,RowIndexEntry> kc;

  kc =ConcurrentLinkedHashCache.create(keyCacheInMemoryCapacity);

 AutoSavingCache<KeyCacheKey,RowIndexEntry> keyCache = new AutoSavingCache<>(kc,CacheType.KEY_CACHE, new KeyCacheSerializer());

int keyCacheKeysToSave =DatabaseDescriptor.getKeyCacheKeysToSave();

keyCache.scheduleSaving(DatabaseDescriptor.getKeyCacheSavePeriod(),keyCacheKeysToSave);

       return keyCache;

   }

分析ConcurrentLinkedHashCache.create(keyCacheInMemoryCapacity)实现过程：

创建了一个：ConcurrentLinkedHashMap<K, V> map;存储所用cache；

具体的创建过程如下：

ConcurrentLinkedHashMap<K, V> map =new ConcurrentLinkedHashMap.Builder<K, V>()

                                           .weigher(entryWeiger)

                                           .maximumWeightedCapacity(weightedCapacity)

                                           .concurrencyLevel(DEFAULT_CONCURENCY_LEVEL)

                                           .build();

d)        initialize keyspaces

        for (String keyspaceName :Schema.instance.getKeyspaces())

        {

            if (logger.isDebugEnabled())

                logger.debug("openingkeyspace {}", keyspaceName);

            // disable auto compaction untilcommit log replay ends

            for (ColumnFamilyStore cfs :Keyspace.open(keyspaceName).getColumnFamilyStores())

            {

                for (ColumnFamilyStore store :cfs.concatWithIndexes())

                {

                   store.disableAutoCompaction();//关闭完以后，关闭自动compaction功能

                }

            }

        }

4)  commlogrecover

代码入口：CommitLog.instance.recover();

    为了保证系统出现异常情况，现在系统选择从系统默认的commitlog恢复日志。这里主要完成这几个操作，发现是否有没有被写到磁盘的数据，恢复这个数据，构建新的日志文件。CommitLog 日志文件的恢复策略是，在头文件中发现没有被序列化的最新的ColumnFamily Id，然后取出这个这个被序列化 RowMutation 对象的起始地址，反序列化成为 RowMutation 对象，后面的操作和新添一条数据的流程是一样的，如果这个 RowMutation 对象中的数据被成功写到磁盘中，那么会在 CommitLog 去掉已经被持久化的 ColumnFamily Id。关于 CommitLog 日志文件的存储格式以及数据如何写到 CommitLog 文件中。

5)   auto compaction

在启动过程中，需要让每个keyspace去compaction，sstable的数据的也将flush到磁盘。所有如果在集群重启以后，这里会提交compact。

具体的实现代码如下：

   if(store.getCompactionStrategy().shouldBeEnabled())

                       store.enableAutoCompaction();

6)  GCInspectorregister

     GCInspector.instance.start 服务。主要是统计统计当前系统中资源的使用情况，将这个信息记录到日志文件中，这个可以作为系统的监控日志使用。

7)  StorageService.instance.initServer()

?  init StorageProxy

Class.forName("org.apache.cassandra.service.StorageProxy");

?  init IndexSummaryManager

Class.forName("org.apache.cassandra.io.sstable.IndexSummaryManager");

?  从系统peers获取该节点的ring和hostid；

?  启动gossipservice，保证可以与其他节点通信；关于gossip怎么样通信，后面会详细分析其通信过程；

?  HintedHandOffManager.instance.start();

?  BatchlogManager.instance.start();

?  MessagingService.instance().listen(FBUtilities.getLocalAddress());

?  LoadBroadcaster.instance.startBroadcasting();

?  Thift  init

?  native transport int

2.  start

启动过程主要包含2个步骤：

1)        nativeServer.start();

2)        thriftServer.start();

   务启动工作已经setup步骤完成；下面专门分析nativeServer的启动过程，nativeServer使用了netty的通信模型，Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。本文第四小节的代码是使用netty

的example，读者感兴趣可以调试。

 nativeserver 具体设置如下：

eventExecutorGroup = newRequestThreadPoolExecutor();

workerGroup= newNioEventLoopGroup();

ServerBootstrapbootstrap

 = new ServerBootstrap().group(workerGroup)

                     .channel(NioServerSocketChannel.class)

                     .childOption(ChannelOption.TCP_NODELAY,true)

                     .childOption(ChannelOption.SO_KEEPALIVE,DatabaseDescriptor.getRpcKeepAlive())

                      .childOption(ChannelOption.ALLOCATOR,CBUtil.allocator)

                      .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,32 * 1024)

                      .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK,8 * 1024);

在运行过程中，会对channel进行操作注册：

protected voidinitChannel(Channel channel) throws Exception

        {

            ChannelPipeline pipeline =channel.pipeline();

            //pipeline.addLast("debug",new LoggingHandler());

           pipeline.addLast("frameDecoder", newFrame.Decoder(server.connectionFactory));

            pipeline.addLast("frameEncoder",frameEncoder);

            pipeline.addLast("frameDecompressor",frameDecompressor);

            pipeline.addLast("frameCompressor",frameCompressor);

           pipeline.addLast("messageDecoder", messageDecoder);

            pipeline.addLast("messageEncoder",messageEncoder);

            pipeline.addLast(server.eventExecutorGroup,"executor", dispatcher);

        }

3.  接受ConcurrentLinkedHashMap数据结构

Cassandra 源码里面详细该数据结构，实现了可以用来实现一个基于LRU策略的缓存。

1)    linkedHashMap

import java.util.HashMap;

import java.util.Iterator;

import java.util.LinkedHashMap;

import java.util.Map;

public class TestLinkedHashMap {

  public static void main(String args[])

  {

   System.out.println("*************************LinkedHashMap*************");

   Map<Integer,String>map = new LinkedHashMap<Integer,String>();

   map.put(6, "apple");

   map.put(3, "banana");

   map.put(2,"pear");

   for (Iterator it = map.keySet().iterator();it.hasNext();)

   {

    Objectkey = it.next();

    System.out.println(key+"="+ map.get(key));

   }

   System.out.println("*************************HashMap*************");

   Map<Integer,String>map1 = new  HashMap<Integer,String>();

   map1.put(6, "apple");

   map1.put(3, "banana");

   map1.put(2,"pear");

   for (Iterator it = map1.keySet().iterator();it.hasNext();)

   {

    Objectkey = it.next();

    System.out.println(key+"="+ map1.get(key));

   }

  }

}

输出：

运行结果如下:

*************************LinkedHashMap*************
6=apple
3=banana
2=pear
*************************HashMap**************************
2=pear
6=apple
3=banana

分析:

l   LinkedHashmap的特点是put进去的对象位置未发生变化,而HashMap会发生变化；

l   LinkedHashMap非线程安全 需要采用google的ConcurrentLinkedHashMap（https://code.google.com/p/concurrentlinkedhashmap/）

l   可以实现last recently used 功能

2)    ConcurrentLinkedHashMap

ConcurrentHashMap的封装，可以用来实现一个基于LRU策略的缓存.

public static voidmain(String[] args) {

ConcurrentLinkedHashMap<Integer,Integer> map = new

ConcurrentLinkedHashMap.Builder<Integer,Integer>().maximumWeightedCapacity(2);

weigher(Weighers.singleton()).build();

map.put(1, 1); 

map.put(2, 2); 

map.put(3, 3); 

System.out.println(map.get(1));//null已经失效了 

System.out.println(map.get(2));

}

3)    整体架构

l  它本质是额外维护了一个双向链表，每次读和写都要改变相应节点的位置，将其移至队列头；

l  什么时候判断容易已经满了，是根据weight。每个元素都有一个weight，每增加一个元素，weight累计；

l  当达到最大值的时候，就需要剔除最少操作的那个元素了，并且触发相关的事件；

cassandra 服务启动流程

标签：cassandra   spark   大数据   

原文地址：http://blog.csdn.net/u011291159/article/details/42008435