Aerospike介绍Aerospike（简称AS）是一个分布式，可扩展的键值存储的NoSQL数据库。T级别大数据高并发的结构化数据存储读写操作达微妙级，99%的响应可在1毫秒内实现采用混合架构，索引存储在内存中，而数据可存储在机械硬盘(HDD)或固态硬盘(SSD) 上（也可存储在内存）AS内部在访问SSD屏蔽了文件系统层级，直接访问地址，保证了数据的读取速度。AS同时支持二级索引与Client聚合，支持简单的sql操作（aql），相比于其他nosql数据库，有一定优势。Aerospike应用场景-广告行业Aerospike作为一个大容量的NoSql解决方案，并未在国内厂中广泛商使用。它适合对容量要求比较大，QPS相对低一些的场景，目前主要集中于互联网广告行业使用（国外）个性化推荐广告应用个性化推荐厂告是建立在了和掌握消费者独特的偏好和习性的基础之上，对消费者的购买需求做出准确的预测或引导，在合适的位置、合适的时间，以合适的形式向消费者呈现与其需求高度吻合的广告，以此来促进用户的消费行为。用户行为日志收集系统收集日志之后推送到ETL做数据的清洗和转换，把ETL过后的数据发送到推荐引擎计算每个消费者的推荐结果，其中推荐逻辑包括规则和算法两部分具体的规则有用户最近浏览、加入购物车、加入收藏等算法则包括商品相似性、用户相似性、文本相似性、图片相似性等算法。把推荐引擎的结果存入Aerospike集群中，并提供给广告投放引擎实时获取实时竞价广告应用当用户浏览一个加入SSP（供应方平台）的站点时，SSP会把此次请求发送到AD EXCHANGE（广告交易平台），然后ADX会把这次请求发送给多家DSP，DSP（需求方平台）根据自身的DMP（数据管理平台），通过对次用户的了解程度进行竞价，最终竞价胜出的DSP获得展现广告的机会。DSP竞价（RTB：实时竞价）胜出的关键是DMP能够根据用户的历史浏览等数据分析和定位用户属性，其中实时竞价广告中非常重要的一个环节就是UserProfile（用户画像）。分别通过HDFS和HBASE对日志进行离线和实时的分析，然后把用户画像的标签(tag : 程序猿、宅男...)结果存入高性能的Nosql数据库Aerospike中，同时把数据备份到异地数据中心。前端广告投放请求通过决策引擎（投放引擎）向用户画像数据库中读取相应的用户画像数据，然后根据竞价算法出价进行竞价。竞价成功之后就可以展现广告了。而在竞价成功之后，具体给用户展现什么样的广告，就是有上面说的个性化推荐广告来完成的。Aerospike与Redis对比Aerospike是NoSQL的数据存储，Redis是缓存Aerospike是多线程的，而 Redis 是单线程的Redis 需要开发人员自己管理分片并提供分片算法用于在各分片之间平衡数据；client： hash 一致性hashcodis ：代理处理shardingRedisCluster： hash槽而 AerospikeDB 可以自动处理相当于分片的工作；在 Redis 中，为了增加吞吐量，需要增加 Redis 分片的数量，并重构分片算法及重新平衡数据，这通常需要停机；而在 AerospikeDB 中，可以动态增加数据卷和吞吐量，无需停机，并且 AerospikeDB 可以自动平衡数据和流量；在 Redis 中，如果需要复制及故障转移功能，则需要开发人员自己在应用程序层同步数据；而在 AerospikeDB 中，只需设置复制因子，然后由 AerospikeDB 完成同步复制操作，保持即时一致性；而且 AerospikeDB 可以透明地完成故障转移；Redis是在内存中运行的 ，AerospikeDB在内存中存储索引，在HDD、SSD中保存数据，也可以在内存中Aerospike架构Aerospike分为三个层次：Client层：对Aerospike Server中的数据进行访问。包括CRUD、批量操作和基于二级索引的查询是一个“智能”客户端，支持C/C++、Java、GoLang、Python、C#、Php、Ruby、JavaScript等绝大部分主流语言。追踪节点感知数据存储位置，当节点启动或停止时立即感知集群配置变化。具有高效性、稳定性和内部的连接池Distribution层：负责管理集群内部数据的平衡分布、备份、容错和不同集群之间的数据同步。主要包含三个模块：Cluster Management Module用于追踪集群节点。关键算法是确定哪些节点是集群的一部分的Paxos-like一致投票过程。Aerospike实现专门的心跳检测（主动与被动），用于监控节点间的连通性。Data Migration Module当有节点添加或移除时，该模块保证数据的重新分布，按照系统配置的复制因子确保每个数据块跨节点和跨数据中心复制。Transaction Processing Module确保读写的一致性与隔离性，写操作先写副本在写主库。该模块包括：Sync/Async Replication（同步/异步复制）：为保证写一致性，在提交数据之前向所有副本传播更新并将结果返回客户端。Proxy （代理）：集群重配置期间客户端可能出现短暂过期，透明代理请求到其他节点。Duplicate Resolution（副本解析）：当集群从活动分区恢复时，解决不同数据副本之间的冲突。Data层：负责数据的存储，Aerospike 属于弱语法的key-value数据库。数据存储模式如下：namespace命名空间，数据存在命名空间中，相当于RDBMS中的database，最多可设置32个。一个namespace可关联多块SSD，一块SSD只关联一个namespace，每个namespace下包含4096个分片set集合，类似于数据库表，一个namespace最多1023个setrecord记录，类似数据库中的一行记录， 采用弱语法(Schema-Less)的方式，bin可随时增加或减少key主键，类型于数据表中的主键bin类似于数据库字段，支持Java基本数据类型：List、Map、Blob, 一个namespace下最多32767个bin索引：Aerospike Index包含主索引（Primary Index）和二级索引（Second Index），索引存储在内存中，并不保存在硬盘里一级索引（Primary-Index）一级索引位于主键(key)，采用hash表与红黑树的混合二级索引（Secondary-Index）二级索引位于非主键上，这允许一对多关系。采用哈希表和B-tree的混合磁盘：Aerospike可以直接访问硬盘中的raw blocks（原始块）直接寻址，并特别优化了Aerospike的最小化读、大块写和并行SSD来增加响应速度和吞吐量。Aerospike的使用Aerospike的安装从网上下载aerospike-server-community-5.0.0.7-el6.tgzwget https://www.aerospike.com/download/server/latest/artifact/el6下载aerospike-server-community的最新版解压aerospike-server-community-5.0.0.7-el6.tgztar -zxvf aerospike-server-community-5.0.0.7-el6.tgz mv aerospike-server-community-5.0.0.7-el6 aerospike-server安装aerospike-server和aerospike-toolscd aerospike-server ./asinstall验证是否安装成功[root@192 aerospike-server]# yum list installed | grep aerospike aerospike-server-community.x86_64 5.0.0.7-1.el6 installed aerospike-tools.x86_64 3.26.2-1.el6 installed #卸载aerospike [root@localhost ~]# rpm -e aerospike-server-community.x86_64 [root@localhost ~]# rpm -e aerospike-tools.x86_64 [root@localhost ~]# rm -rf /etc/aerospike/aerospike-server启动、停止、重启、状态systemctl start aerospike systemctl stop aerospike systemctl restart aerospike systemctl status aerospikeaerospike-server管理：asadmasadm #进入管理端 Admin> info Admin> i netaerospike-server操作 ：aqlaql aql> show namespaces +------------+ | namespaces | +------------+ | "test" | | "bar" |Aerospike的基本概念namespace 策略容器，类似RDBMS中的database，可以设置副本数、内存大小、有效时长、存储引擎、文件存储位置。Sets 类似RDBMS中的表。Records 类似RDBMS中的行Bin 类似RDBMS中的列，一行可以有多个bins。AerospikeMysqlnamespacedatabasesettablebin（kv中v的位置）columnrecordrowkey（也是pk，kv中k的位置）pkAerospike的数据操作 -- 主键 bins 插入可以不同 INSERT INTO <ns>[.<set>] (PK, <bins>) VALUES (<key>, <values>) DELETE FROM <ns>[.<set>] WHERE PK = <key> <ns> is the namespace for the record. <set> is the set name for the record. <key> is the record's primary key. <bins> is a comma-separated list of bin names. <values> is comma-separated list of bin values -- 没有update -- 当insert 同一pk时，数据为修改 Examples: INSERT INTO test.demo (PK, foo, bar) VALUES ('key1', 123, 'abc') DELETE FROM test.demo WHERE PK = 'key1' insert into test.user1 (PK,name,age,sex,address) VALUES (2,'zhaoyun',21,'M','beijing') insert into test.user2(pk,name,sex,age) values(1,'zhangfei','M',23) -- pk都是1 则是对原纪录的修改 insert into test.user2(pk,name,sex,age) values(1,'diaochan','F',18) QUERY SELECT <bins> FROM <ns>[.<set>] SELECT <bins> FROM <ns>[.<set>] WHERE <bin> = <value> SELECT <bins> FROM <ns>[.<set>] WHERE <bin> BETWEEN <lower> AND <upper> SELECT <bins> FROM <ns>[.<set>] WHERE PK = <key> SELECT <bins> FROM <ns>[.<set>] IN <indextype> WHERE <bin> = <value> SELECT <bins> FROM <ns>[.<set>] IN <indextype> WHERE <bin> BETWEEN <lower> AND <upper> <ns> is the namespace for the records to be queried. <set> is the set name for the record to be queried. <key> is the record's primary key. <bin> is the name of a bin. <value> is the value of a bin. <indextype> is the type of a index user wants to query. (LIST/MAPKEYS/MAPVALUES) <bins> can be either a wildcard (*) or a comma-separated list of bin names. <lower> is the lower bound for a numeric range query. <upper> is the lower bound for a numeric range query. Examples: SELECT * FROM test.demo SELECT * FROM test.demo WHERE PK = 'key1' SELECT foo, bar FROM test.demo WHERE PK = 'key1' SELECT foo, bar FROM test.demo WHERE foo = 123 SELECT foo, bar FROM test.demo WHERE foo BETWEEN 0 AND 999 select * from test.user2 where name='zhaoyun' -- 没有建立索引 ，不能查询 Error: (201) AEROSPIKE_ERR_INDEX_NOT_FOUND create index idx_1 on test.user2(name) string select * from test.user2 where name='zhaoyun' +-----------+-----+-----+-----------+ | name | sex | age | address | +-----------+-----+-----+-----------+ | "zhaoyun" | "M" | 21 | "beijing" | +-----------+-----+-----+-----------+ CREATE INDEX <index> ON <ns>[.<set>] (<bin>) NUMERIC|STRING|GEO2DSPHERE CREATE LIST/MAPKEYS/MAPVALUES INDEX <index> ON <ns>[.<set>] (<bin>) NUMERIC|STRING|GEO2DSPHERE CREATE INDEX idx_foo ON test.demo (foo) NUMERIC DROP INDEX test.demo idx_fooAerospike的客户端（Java）pom.xml引入aerospike-client<dependency> <groupId>com.aerospike</groupId> <artifactId>aerospike-client</artifactId> <version>4.4.9</version> </dependency>aerospike-client的API使用//IP+port AerospikeClient client=new AerospikeClient("192.168.127.128",3000); //写策略 WritePolicy wp=new WritePolicy(); //超时时间 wp.setTimeout(1000); //key Key k1=new Key("test","user1",1); //bins // KV Bin b11=new Bin("name","zhangfei"); Bin b12=new Bin("sex","M"); Bin b13=new Bin("age",23); //写值 client.put(wp,k1,b11,b12,b13); //读值 Record r1=client.get(wp,k1,"name","age","sex"); System.out.println(r1); System.out.println("==================================="); Key k2=new Key("test","user1",2); //bins // KV Bin b21=new Bin("name","diaochan"); Bin b22=new Bin("sex","F"); Bin b23=new Bin("age",21); //写值 client.put(wp,k2,b21,b22,b23); //取得指定key的数据 //批量执行策略 BatchPolicy bp=new BatchPolicy(wp); //key的数组 Key[] ks={k1,k2}; //循环输出 for(Record r:client.get(bp,ks)){ System.out.println(r); }Aerospike集群实现Aerospike集群管理集群处理节点成员身份，并确保当前成员和所有集群中的节点保持一致。包括：集群视图、节点发现和视图改变集群视图每个Aerospike节点都会自动分配唯一的节点标识符，是由MAC地址和监听端口组成的。包括：cluster_key：是一个随机生成的8字节值，用于标识集群视图的实例。succession_list:是作为集群一部分的唯一节点标识符集合。节点发现节点间通过心跳消息来检测节点的有效或失效集群中的每个节点维护一个邻接表，是最近向该节点发送心跳消息的其他节点列表如果心跳超时，则表示该节点失效，从邻接表中删除Aerospike还采用其他信息作为备用二次心跳机制集群中的每个节点通过计算平均消息丢失来评估每个相邻节点的运行状态评分视图改变领接表一旦发生改变，就会触发运行一个Paxos共识算法来确定一个新的集群视图。Aerospike把邻接表中节点标识符最高的节点当Proposer，并承担Proposal的角色Proposal提成新的集群视图，如果被接受，则节点开始重新分配数据（Rebalence）Aerospike数据分布Aerospike有智能分区算法，即把用户输入的key在内部根据RIPEMD-160算法，重新hash出一个key并取前20位，然后相对均衡的把数据分布到各个节点之上。并且满足namespace配置文件的配置（例如保存多少个备份、是存在磁盘还是存在内存中）。每个Digest Space（namespace）被分为4096个不重叠的分区（partitions），它是Aerospike数据存储的最小单位如上，一个4个节点的集群，每个节点存储1/4数据的主节点，同时也存储1/4数据的副本。如果节点1不可访问，节点1的副本将被拷贝到其他节点上。复制因子（ replication factor）是一个配置参数，不能超过集群节点数。副本越多可靠性越高。作为必须经过所有数据副本的写请求也越高。实践中，大部分部署使用的数据因子为2（一份主数据和一个副本）。 同步复制保证即时一致性，没有数据丢失。在提交数据并返回结果给客户端之前，写事务被传播到所有副本。主成功同时备成功后，客户端认为是成功在集群重新配置期间，当Aerospike智能终端发送请求到那些短暂过时的错误节点时，Aerospike智能集群会透明的代理请求至正确的节点。Aerospike集群配置和部署有两种方式可以搭建集群：Multicast组播方式（UDP）和Mesh网格方式（TCP）Multicast组播方式（UDP）heartbeat { mode multicast multicast-group 239.1.139.1 port 3000 address 192.168.127.131 interval 150 timeout 10 }udp是不可靠协议，所以有可能会造成节点脱落，而且网络有可能不支持组播Mesh网格方式（TCP）heartbeat { mode mesh # add current node address here address 192.168.127.131 port 3000 # add all cluster node address here mesh-seed-address-port 192.168.127.131 3002 mesh-seed-address-port 192.168.127.128 3002 interval 150 timeout 10 }集群部署由于是rpm的安装方式，所以最好每个虚拟机只装一个Aerospike在192.168.127.128上安装Aerospike后，修改配置文件/etc/aerospike/aerospike.confvim /etc/aerospike/aerospike.confservice { user root group root paxos-single-replica-limit 1 # Number of nodes where the replica count is automatically reduced to 1. pidfile /var/run/aerospike/asd.pid proto-fd-max 15000 } logging { # Log file must be an absolute path. file /var/log/aerospike/aerospike.log { context any info } } network { service { address any port 3000 access-address 192.168.127.128 3002 } heartbeat { mode mesh address 192.168.127.128 port 3002 #all cluster mesh-seed-address-port 192.168.127.128 3002 mesh-seed-address-port 192.168.127.131 3002 # To use unicast-mesh heartbeats, remove the 3 lines above, and see # aerospike_mesh.conf for alternative. interval 150 timeout 10 } fabric { address any port 3001 } info { address any port 3003 } } namespace test { replication-factor 2 memory-size 256M storage-engine memory } namespace bar { replication-factor 2 memory-size 256M storage-engine memory }在192.168.127.131上安装Aerospike后，修改配置文件service { user root group root paxos-single-replica-limit 1 # Number of nodes where the replica count is automatically reduced to 1. pidfile /var/run/aerospike/asd.pid proto-fd-max 15000 } logging { # Log file must be an absolute path. file /var/log/aerospike/aerospike.log { context any info } } network { service { address any port 3000 access-address 192.168.127.131 3002 } heartbeat { mode mesh address 192.168.127.131 port 3002 #all cluster mesh-seed-address-port 192.168.127.131 3002 mesh-seed-address-port 192.168.127.128 3002 # To use unicast-mesh heartbeats, remove the 3 lines above, and see # aerospike_mesh.conf for alternative. interval 150 timeout 10 } fabric { address any port 3001 } info { address any port 3003 } } namespace test { replication-factor 2 memory-size 256M storage-engine memory } namespace bar { replication-factor 2 memory-size 256M storage-engine memory }Aerospike集群的访问Host[] hosts = new Host[]{ new Host("192.168.127.128", 3000), new Host("192.168.127.131", 3000) }; ClientPolicy policy = new ClientPolicy(); AerospikeClient client = new AerospikeClient(policy, hosts); //写策略 WritePolicy wp = new WritePolicy(); //超时时间 wp.setTimeout(500); Key key1 = new Key("test", "SUser", "11"); Bin bin11 = new Bin("name", "zhangfei-c"); Bin bin12 = new Bin("age", 25); Bin bin13 = new Bin("sex", "M-c"); client.put(wp, key1, bin11, bin12, bin13); Key key2 = new Key("test", "SUser", "22"); Bin bin21 = new Bin("name", "zhaoyun-c"); Bin bin22 = new Bin("age", 24); Bin bin23 = new Bin("sex", "M-c"); client.put(wp, key2, bin21, bin22, bin23); Record r1 = client.get(wp, key1, "name", "age", "sex"); System.out.println(r1);
Tair介绍Tair简介Tair(Taobao Pair)是淘宝开发的分布式Key-Value存储引擎，服务器端自动负载均衡，分为持久化和非持久化两种方式存储非持久化：分布式缓存使用 Memcached（mdb）、Redis（rdb）持久化：SQL-DB使用FireBird（fdb）NoSQL-DB：使用Kyoto Cabinet（kdb）、LevelDB（ldb）Tair采用可插拔存储引擎设计，以上这些存储引擎可以很方便的替换，还可以引入新的存储引擎比如：MySQL使用场景分布式缓存大多数使用场景：大访问少量临时数据的存储（kb左右）用于缓存，降低对后端数据库的访问压力session场景：高速访问某些数据结构的应用和计算（rdb）数据源存储快速读取数据（fdb）持续大数据量的存入读取（ldb），交易快照高频度的更新读取（ldb），库存Tair整体架构分析一个Tair集群主要包括client、Config server和Dataserver 三个不同的应用。Client在初始化时，从Config server处获取数据的分布信息，根据分布信息和相应的Data server交互完成用户的请求。Config Server通过和Data Server的心跳（HeartBeat）维护集群中可用的节点，并根据可用的节点，构建数据的在集群中的分布信息Data server负责数据的存储，并按照Config server的指示完成数据的复制和迁移工作。Config ServerConfig Server是单点，采用一主一备的方式保证可靠性。管理所有的data server，维护data server的状态信息，用户配置的桶数量、副本数、机房信息和数据分布的对照表协调数据迁移、管理进度，将数据迁移到负载较小的节点上Client和ConfigServer的交互主要是为了获取数据分布的对照表，当client获取到对照表后，会cache这张表，然后通过查这张表决定数据存储的节点，所以不需要和configserver交互，这使得Tair对外的服务不依赖configserver，所以它不是传统意义上的中心节点Config server维护的对照表有版本概念，由于集群变动或管理触发，构建新的对照表后，对照表的版本号递增，并通过Data server的心跳，将新表同步给数据节点。客户端和Data server交互时，Dataserver每次都把自己缓存的对照表版本号放入response结构中，返回给客户端，客户端将Data server的对照表版本号和自己缓存的对照表版本号比较，如果不相同，会主动和Config server通信，请求新的对照表。Tair的Configserver使客户端使用时候，不需要配置数据节点列表，也不需要处理节点的的状态变化，这使得Tair对最终用户来说使用和配置都很简单。Data ServerData server负责数据的物理存储，并根据Configserver构建的对照表完成数据的复制和迁移工作。Data server具备抽象的存储引擎层，可以很方便地添加新存储引擎。Data server还有一个插件容器，可以动态地加载/卸载插件。Tair的存储引擎有一个抽象层，只要满足存储引擎需要的接口，便可以很方便地替换Tair底层的存储引擎。比如你可以很方便地将bdb、tc甚至MySQL作为Tair的存储引擎，而同时使用Tair的分布方式、同步等特性。Tair默认包含两个存储引擎：mdb和fdb。mdb是一个高效的缓存存储引擎，它有着和memcached类似的内存管理方式。mdb支持使用share memory（tmpfs），这使得我们在重启Tair数据节点的进程时不会导致数据的丢失，从而使升级对应用来说更平滑，不会导致命中率的较大波动。fdb是一个简单高效的持久化存储引擎，使用树的方式根据数据key的hash值索引数据，加快查找速度。索引文件和数据文件分离，尽量保持索引文件在内存中，以便减小IO开销。使用空闲空间池管理被删除的空间。Tair的安装与使用环境准备Gityum install git yum update -y nss curl libcurlsvnyum install subversionLibtoolyum install libtoolboost-develyum install boost-develzlibyum install zlib-develC++yum install gcc-c++项目源码下载和编译tbsys库和tbnet库下载（tair底层依赖tbsys库和tbnet库）#git源码下载 git clone https://github.com/kayaklee/tb-common-util.git #设置环境变量 mkdir /var/tblib export TBLIB_ROOT="/var/tblib" #因为tbnet和tbsys在两个不同的目录，但它们的源码文件里头文件的互相引用却没有加绝对或相对路径，将两个目录的源码加入到C++环境变量中即可。否则编译时会出现：“fatal error:tysys.h: No such file or directory”的错误。 CPLUS_INCLUDE_PATH=/home/tair/tb-common-util/tbsys/src:/home/tair/tb-commonutil/tbnet/src export CPLUS_INCLUDE_PATH #修改tbsys代码 cd ~/tb-common-util/trunk/tbsys/src #下载的代码有个错误：具体是tbsys/src/tblog.cpp中323行代码:需要将CLogger::CLogger&CLogger::getLogger()改为CLogger& CLogger::getLogger() #编译tbsys和tbnet cd tb-common-utils ./build.sh #注：安装成功后，TBLIB_ROOT所指示的目录下会有include和lib两个目录。tair下载#git源码下载 git clone https://github.com/alibaba/tair.git #安装依赖 yum install -y openssl-devel libcurl-devel #编译 ./bootstrap.sh #检测和生成 Makefile ./configure #编译和安装到目标目录 make -j make install #注：默认安装位置是 ~/tair_binTair配置和启动下面以MDB引擎为例配置一个最小化的Tair集群（1 ConfigServer + 1 DataServer）#查看和设置系统tmpfs #MDB 引擎默认使用共享内存，所以需要查看并设置系统的tmpfs的大小 # 这里根据实际机器内存情况配置，必须大于Tair使用内存的配置 vim /etc/fstab tmpfs /dev/shm tmpfs defaults,size=1024M 0 0 #生效 mount -o remount /dev/shm #查看tmpfs cat /etc/fstab | grep /dev/shm #显示 tmpfs /dev/shm tmpfs rw,size=1G 0 0 # 定义配置文件 复制配置文件 cp etc/configserver.conf.default etc/configserver.conf cp etc/group.conf.default etc/group.conf etc/dataserver.conf.default etc/dataserver.confconfigserver.conf[public] config_server=192.168.127.133:5198 #config_server=192.168.1.2:5198 #config_server=10.211.55.9:5198 #config_server=192.168.1.2:5198 dev_name=eno16777736group.conf# data center A _server_list=192.168.127.133:5191 #_server_list=192.168.1.2:5191 #_server_list=192.168.1.3:5191 #_server_list=192.168.1.4:5191 # data center B #_server_list=192.168.2.1:5191 #_server_list=192.168.2.2:5191 #_server_list=192.168.2.3:5191 #_server_list=192.168.2.4:5191dataserver.conf[public] config_server=192.168.127.133:5198 #config_server=192.168.1.2:5198 dev_name=eno16777736 mdb_inst_shift=0 process_thread_num=4 io_thread_num=4 #这里 slab_mem_size控制MDB内存池的总大小，mdb_inst_shift 控制实例的个数，注意这里一个实例必须大于512MB且小于64GB。 slab_mem_size=512修改tair.sh启动脚本#在CentOS 7下，安装目录下的 tair.sh 启动脚本有一行代码需要修改 tmpfs_size=`df -m |grep /dev/shm | awk '{print $2}'`启动Tair实例#启动dataserver tair_bin $ ./tair.sh start_ds #启动configserver tair_bin $ ./tair.sh start_cs #查看进程 ps -ef |grep tair #注：执行后没有两行记录显示，说明启动失败，可查看logs中的log查看错误信息。Tair测试#客户端读写测试 tair_bin $ ./sbin/tairclient -c 192.168.127.133:5198 -g group_1 TAIR> health TAIR> put key value TAIR> get key TAIR> remove key TAIR> get keyTair停止#停止dataserver tair_bin $ ./tair.sh stop_ds #停止configserver tair_bin $ ./tair.sh stop_csTair高可用和负载均衡Tair的高可用和负载均衡，主要通过对照表和数据迁移两大功能进行支撑。对照表将数据分为若干个桶，并根据机器数量、机器位置进行负载均衡和副本放置，确保数据分布均匀，并且在多机房有数据副本。在集群发生变化时，会重新计算对照表，并进行数据迁移。对照表在Tair系统中，采用对照表将数据均衡的分布在DataServer上，还能动态适应节点的扩容和缩容Tair基于一致性Hash算法存储数据，根据配置建立固定数量的桶（bucket），桶为Hash环节点，hash(key) 顺时针 设置桶桶是负载均衡和数据迁移的基本单位config server 根据一定的策略把每个桶指派到不同的data server上，因为数据按照key做hash算法，所以可以认为每个桶中的数据基本是平衡的，保证了桶分布的均衡性，就保证了数据分布的均衡性。比如：bucket data server 0 192.168.127.132 192.168.127.134 1 192.168.127.133 192.168.127.135 2 192.168.127.132 192.168.127.134 3 192.168.127.133 192.168.127.135 4 192.168.127.132 192.168.127.134 5 192.168.127.133 192.168.127.135Tair支持自定义的备份数，比如你可以设置数据备份为2，以提高数据的可靠性。对照表可以很方便地支持这个特性。第二列为主节点的信息，第三列为辅节点信息。在Tair中，客户端的读写请求都是和主节点交互。当有节点不可用时，如果是辅节点，那么configserver会重新为其指定一个辅节点，如果是持久化存储，还将复制数据到新的辅节点上。如果是主节点，那么configserver首先将辅节点提升为主节点，对外提供服务，并指定一个新的辅节点，确保数据的备份数。对照表的初始化第一次启动ConfigServer会根据在线的DataServer生成对照表ConfigServer会启动一个线程（table_builder_thread），该线程会每秒检查一次是否需要重新构造对照表Tair 提供了两种生成对照表的策略：负载均衡优先config server会尽量的把桶均匀的分布到各个data server上一个桶的备份数据不能在同一台主机上位置安全优先一般我们通过控制 _pos_mask（Tair的一个配置项） 来使得不同的机房具有不同的位置信息一个桶的备份数据不能都位于相同的一个位置（不在同一个机房）数据迁移当DataServer增加时，config server负责重新计算一张新的桶在data server上的分布表，将桶较为平均的分配到各个DataServer上当DataServer发生故障时，config server负责重新计算一张新的桶在data server上的分布表，将原来由故障机器服务的桶的访问重新指派到其它的data server中这时需要做数据迁移，具体流程如下：设置当前正在迁移的桶IDDataServer写入桶数据时，会写入redologmigrate_manager迁移内存中的桶数据migrate_manager迁移redolog数据redolog数据迁移完成后，将桶标记为迁移完成将信息发送给ConfigServer用于同步对照表Tair存储引擎Tair的存储引擎有一个抽象层（storage_manager）,只要实现存储引擎接口，便可以替换Tair的底层存储引擎。可插拔存储引擎---类似MySQLTair默认包含四种存储引擎：mdb、fdb、kdb和ldbmdb数据存储：Memcached高效缓存存储使用share memory，重启不会数据丢失支持K-V存储、prefix操作适用于：String缓存使用 （json）、大访问少量的临时数据存储、Session分离fdb数据存储：FireBird高效的持久化SQL存储索引文件和数据文件分离------>mysql MyISam使用Tree的方式根据key的hash值索引数据 B tree索引文件在内存中适用于：快速访问较小的数据kdb数据存储：Kyoto CabinetCabinet开发的KV的持久化存储简单的包含记录的数据文件存储形式为hash表或B+Tree适用于：简单临时存储ldb数据存储：LevelDB是Google开发的高性能持久化KV存储可内嵌mdb缓存 mdb+ldb支持kv、prefix支持批量操作适用于：大数据量的存取（交易快照）、高频度的更新（库存）、离线大批量数据导入rdb数据存储：Redis高效缓存存储多种数据结构和计算适用于：复杂数据结构存储、商品属性、粉丝列表、商品评论、消息队列 数据存储持久化数据格式应用场景mdbMemcached否非K-V存储、prefix操作缓存使用、Session分离fdbFireBird是SQL、BTree快速访问较小的数据kdbKyoto Cabinet是Hash、B+Tree简单临时存储ldbLevelDB是kv、prefix、batch大数据量高频度的存储rdbRedis否List、Set、Hash....复杂数据结构缓存阿里使用：mdb、rdb、ldbmdb的存储结构（Memcached）mem_pool用于共享内存管理，将内存分为若干个pagepage个数根据slab_mem_size设置，单位为MB单个DataServer最多使用64G内存mem_cache用于管理slab，存放slab_manager列表slab_manager管理item(数据块)slab个数为100，一个slab可存储800kbcache_hash_map用于存储hash表，根据key进行hash对应item数据hash冲突，产生链表mdb_area_stat用于维护area（namespace）状态，记录了该area的数据链表和数据量限制(tablespace)Tair相应APIkey/value常见操作API 操作说明getget key [area]获得key的值putput key data [area] [expired]设置key和valueincrincr key [count] [initValue] [area]对key的值自增decrdecr key [count] [area]对key的值自减batcheGetmget key1 ... keyn area批量获得expireexpire key time设置key的过期时间prefixPutpput [area] pkey skey value根据前缀设置，按照prefix计算hash，同一个prefix会存储在同一个hash表，形成链表prefixGetpget [area] pkey skey根据前缀读取versionTair中的每个数据都包含版本号，版本号在每次更新后都会递增。这个特性有助于防止由于数据的并发更新导致的问题，类似于乐观锁比如，系统有一个value为”a“，A和B同时get到这个value。A执行操作改为"b"B执行操作改为"c"如果不控制，无论A和B谁先更新成功，它的更新都会被后到的更新覆盖。使用version解决第一次put version为1get version为1A修改成功后 version增加1B修改时本身version为1，小于服务器版本2所以B修改不成功在put时，version传0 可强制修改struct mdb_item{ uint64_t item_id; ..... uint16_t version }Tair实现乐观锁trylock（获取版本）–> transaction –> unlock(版本检查) — 正常—> version+1 commit — 异常—> rollback只要能获取版本，就认为trylock成功版本检查则更新version，一般是加1；否则，异常处理，version不变，并开启回滚多事务同时获得乐观锁时，unlock只能有一个成功，其余的都应该失败version 初始化要大于1Tair实现分布式锁利用 Tair 的 version 特性可以实现分布式锁，由于 LDB 具有持久化功能，当服务有出现宕机的情况，也不会因此出现锁丢失或者锁不可释放的情况。如果KEY不存在的话，传入一个固定的初始化VERSION（需要大于1），Tair会在保存这个缓存的同时设置这个缓存的VERSION为你传入的 VERSION+1KEY如果已经存在，Tair会校验你传入的VERSION是否等于现在这个缓存的VERSION，如果相等则允许修改，否则将失败。示例代码如下：//获得锁 public Boolean lock(String lockKey) { // 5 :version expiretime : 过期时间 秒 ResultCode code = defaultTairManager.put(lockKey, defaultVlaue, 5, expiretime); if (ResultCode.SUCCESS.equals(code)) return true; else return false; } //释放锁 public Boolean unlock(String lockKey){ ResultCode code = defaultTairManager.delete(lockKey); return ResultCode.SUCCESS.equals(code); }Java Clientpom.xml<dependency> <groupId>com.taobao.tair</groupId> <artifactId>tair-client</artifactId> <version>2.3.5</version> </dependency>demoDefaultTairManager defaultTairManager = new DefaultTairManager(); List<String> cs = new ArrayList<String>(); cs.add("192.168.127.132:5198"); defaultTairManager.setConfigServerList(cs); defaultTairManager.setGroupName("group_1"); defaultTairManager.init(); defaultTairManager.put(0,"name:001","zhangfei"); Result<DataEntry> value= defaultTairManager.get(0,"name:001"); System.out.println(value); //设置版本0 强制更新 过期时间2秒 defaultTairManager.put(0,"name:002","zhaoy",0,2); Result<DataEntry> value2= defaultTairManager.get(0,"name:002"); System.out.println(value2);Tair机房、负载均衡、数据迁移、不停止服务更好的可用性和负载均衡Redis多数据结构处理、计算可以使用Tair+Redis
EVCache介绍EVCache是一个开源、快速的分布式缓存，是基于Memcached的内存存储和Spymemcached客户端实现的，是Netflix（网飞）公司开发的E：Ephemeral：数据存储是短暂的，有自身的存活时间V：Volatile：数据可以在任何时候消失Cache：内存级键值对存储Rend服务：是一个代理服务，用GO语言编写，能够高性能的处理并发。Memcached：基于内存的键值对缓存服务器Mnemonic：基于硬盘（SSD）的嵌入式键值对存储服务器，封装了RocksDB(是一种SSD的技术)EVCache集群在峰值每秒可以处理200kb的请求，Netflix生产系统中部署的EVCache经常要处理超过每秒3000万个请求，存储数十亿个对象，跨数千台memcached服务器。整个EVCache集群每天处理近2万亿个请求。EVCache集群响应平均延时大约是1-5毫秒，最多不会超过20毫秒。EVCache集群的缓存命中率在99%左右。EVCache的应用典型用例Netflix用来构建超大容量、高性能、低延时、跨区域的全球可用的缓存数据层EVCache典型地适合对强一致性没有必须要求的场合典型用例：Netflix向用户推荐用户感兴趣的电影典型部署EVCache 是线性扩展的，可以在一分钟之内完成扩容，在几分钟之内完成负载均衡和缓存预热。单节点部署集群启动时，EVCache向服务注册中心（Zookeeper、Eureka）注册各个实例在web应用启动时，通过EVCache的客户端(EVCache Client)查询命名服务中的EVCache服务器列表，并建立连接。web应用通过EVCache的客户端操作数据，客户端通过key使用一致性hash算法，将数据分片到集群上。多可用区部署EVCache的跨可用区复制：EVCache客户端库包发送SET到缓存系统的本地地区的一个实例服务器中。EVCache客户端库包同时也将写入元数据(包括key，但是不包括要缓存的数据本身)到复制消息队列(Kafka)本地区的“复制转播”的服务将会从这个消息队列中读取消息。转播服务会从本地缓存中抓取符合key的数据转播会发送一个SET请求到远地区的“复制代理”服务。在远地区，复制代理服务会接受到请求，然后执行一个SET到它的本地缓存，完成复制。在接受地区的本地应用当通过GET操作以后会在本地缓存看到这个已经更新的数据值。EVCache的安装与使用由于Netflix没有开源EVCache的服务器部分，这里采用Memcached作为服务器。安装Memcached#安装libevent库 yum install libevent libevent-devel gcc-c++ #下载最新的memcached wget http://memcached.org/latest #解压 tar -zxvf latest #进入目录 cd memcached-1.6.6 #配置 ./configure --prefix=/usr/memcached #编译 make #安装 make install #启动 memcached -d -m 1000 -u root -l 192.168.127.131 -p 11211 -c 256 -P /tmp/memcached.pid-d 选项是启动一个守护进程-m 是分配给Memcache使用的内存数量，单位是MB，我这里是10MB-u 是运行Memcache的用户，我这里是root-l 是监听的服务器IP地址，我这里指定了服务器的IP地址192.168.127.131-p 是设置Memcache监听的端口，我这里设置了11211（默认），最好是1024以上的端口-c 选项是最大运行的并发连接数，默认是1024，我这里设置了256，按照你服务器的负载量来设定-P 是设置保存Memcache的pid文件，我这里是保存在 /tmp/memcached.pid使用EVCache Clientpom<dependency> <groupId>com.netflix.evcache</groupId> <artifactId>evcache-client</artifactId> <version>4.139.0</version> </dependency> <dependency> <groupId>net.spy</groupId> <artifactId>spymemcached</artifactId> <version>2.12.3</version> </dependency> <dependency> <groupId>com.netflix.eureka</groupId> <artifactId>eureka-client</artifactId> <version>1.5.6</version> <scope>runtime</scope> </dependency> <dependency> <groupId>com.netflix.spectator</groupId> <artifactId>spectator-nflx-plugin</artifactId> <version>0.80.1</version> <scope>runtime</scope> </dependency> <dependency> <groupId>com.netflix.spectator</groupId> <artifactId>spectator-api</artifactId> <version>0.80.1</version> </dependency> <dependency> <groupId>com.netflix.rxjava</groupId> <artifactId>rxjava-core</artifactId> <version>0.20.7</version> </dependency> <dependency> <groupId>com.netflix.servo</groupId> <artifactId>servo-core</artifactId> <version>0.12.25</version> </dependency> <dependency> <groupId>com.google.code.findbugs</groupId> <artifactId>annotations</artifactId> <version>3.0.1</version> </dependency> <dependency> <groupId>com.netflix.nebula</groupId> <artifactId>nebula-core</artifactId> <version>4.0.1</version> <scope>runtime</scope> </dependency> <dependency> <groupId>com.netflix.archaius</groupId> <artifactId>archaius2-core</artifactId> <version>2.3.13</version> <scope>runtime</scope> </dependency> <dependency> <groupId>com.netflix.archaius</groupId> <artifactId>archaius-aws</artifactId> <version>0.6.0</version> </dependency> <dependency> <groupId>javax.inject</groupId> <artifactId>javax.inject</artifactId> <version>1</version> </dependency> <dependency> <groupId>io.reactivex</groupId> <artifactId>rxjava</artifactId> <version>1.3.8</version> </dependency>编写EVCache代码String deploymentDescriptor = System.getenv("EVCACHE_SERVER"); if ( deploymentDescriptor == null ) { deploymentDescriptor = "SERVERGROUP1=192.168.127.131:11211"; } System.setProperty("EVCACHE_1.use.simple.node.list.provider", "true"); System.setProperty("EVCACHE_1-NODES", deploymentDescriptor); EVCache evCache = new EVCache.Builder().setAppName("EVCACHE_1").build(); // s:key t :value i:ttl evCache.set("name","zhangfei",10); String v = evCache.get("name"); System.out.println(v);EVCache原理EVCache的内存存储是基于Memcached实现的EVCache的客户端是基于Spymemcached实现的Memcached内存存储Memcached简介Memcached是danga（丹加）开发的一套分布式内存对象缓存系统，用于在动态系统中减少数据库负载，提升性能Memcached是C/S模式的，基于libevent的事件处理Libevent 是一个用C语言开发的，高性能;轻量级，专注于网络，不如 ACE 那么臃肿庞大；源代码相当精炼、易读；跨平台，支持 Windows、 Linux、 BSD 和 Mac OS；支持多种 I/O 多路复用技术，epoll、 poll、 select 和 kqueue 等；支持 I/O，定时器和信号等事件；注册事件优先级。Memcached是多线程的 MemcachedRedis多线程是否数据结构简单kv，V为一个值V多种，List、Set、Hash....持久化否是集群客户端或代理端（twemproxy）客户端、代理端(codis)、服务器端（RedisCluster）性能读多写少，大数据 大于100K读写都行通信协议文本（telnet）、二进制RESP集群内通信无Gossip、订阅数据安全无有集群failover无有Slab Allocation机制传统的内存分配是通过对所有记录简单地进行malloc（动态内存分配）和free（释放） 来进行的。是以Page（M）为存储单位的(BuddySystem)。这种方式会导致内存碎片，加重操作系统内存管理器的负担。memcached采用 Slab（块） Allocation的方式分配和管理内存slab是Linux操作系统的一种内存分配机制slab分配器分配内存以Byte为单位，专为小内存分配而生Slab Allocation的原理：根据预先设定的大小(Page=1M)， memcached -m 分配的内存 默认64M将分配的内存分割成各种大小的块(Chunk)，并把尺寸相同的块分成组(Slab Class)，Memcached根据收到的数据大小，选择最合适的slabClass进行存储注：块越小存的数据越多，块越来越大，是由growth factor决定的（1.25）数据ItemItem就是我们要存储的数据。是以双向链表的形式存储的。/** * Structure for storing items within memcached. */ typedef struct _stritem { struct _stritem *next; //next即后向指针 struct _stritem *prev; //prev为前向指针 struct _stritem *h_next; /* hash chain next */ rel_time_t time; /* least recent access */ rel_time_t exptime; /* expire time */ int nbytes; /* size of data */ unsigned short refcount; uint8_t nsuffix; /* length of flags-and-length string */ uint8_t it_flags; /* ITEM_* above */ uint8_t slabs_clsid;/* which slab class we're in */ uint8_t nkey; /* key length, w/terminating null and padding */ union { uint64_t cas; char end; } data[]; } item;item 的结构分两部分:item结构定义next、prev、time(最近访问时间)、exptime（过期的时间）、nkey（key的长度）、refcount（引用次数）、nbytes（数据大小）、slabs_clsid（从哪个 slabclass 分配而来）item数据： CAS，key，suffix，value 组成缓存过期机制Memcached有两种过期机制：Lazy Expiration（惰性过期）和LRULazy ExpirationMemcached在get数据时，会查看exptime，根据当前时间计算是否过期(now-exptime>0)，如果过期则删除该数据LRU当Memcached使用内存大于设置的最大内存(-m 启动指定 默认64M)使用时，Memcached会启动LRU算法淘汰旧的数据项。使用slabs_alloc函数申请内存失败时，就开始淘汰数据了。淘汰规则是，从数据项列表尾部开始遍历，在列表中查找一个引用计数器(refcount)为0的item，把此item释放掉。如果在item列表找不到计数器为0的item，就查找一个3小时没有访问过的item(now-time>3H)。把他释放，如果还是找不到，就返回NULL（申请内存失败）。当内存不足时，memcached会把访问比较少或者一段时间没有访问的item淘汰，以便腾出内存空间存放新的item。Spymemcached设计思想spymemcached 是一个 memcached的客户端，使用NIO实现。memcached的客户端有以下几种：memcachedclient danga公司spymemcached spy公司xmemcached主要有以下特性：Memcached协议支持Text和Binary(二进制)异步通信：使用NIO，采用callback集群：支持sharding机制自动恢复：断网重连failover：可扩展容错，支持故障转移支持批量get、支持jdk序列化整体设计API接口：提供同步或异步接口调用，异步接口返回Future任务封装：将访问的操作及callback封装成Task路由分区：通过Sharding策略（后面讲），选择Key对应的连接（connection）将Task放到对应连接的队列中Selector异步获取队列中的Task，进行协议的封装和发送相应Memcached收到Memcached返回的包，会找到对应的Task，回调callback，进行协议解析并通知业务线程Future.getAPI接口设计对外API接口有两种，同步和异步同步接口：比如set、get等异步接口：asyncGet、asyncIncr等//获取一个连接到几个服务端的memcached的客户端 MemcachedClient c = new MemcachedClient(AddrUtil.getAddresses("192.168.127.123:11211")); //获取值，如果在5秒内没有返回值，将取消 Object myObj = null; Future<Object> f = c.asyncGet("name"); try{ //异步获取 阻塞 myObj = f.get(5,TimeUnit.SECONDS); } catch(TimeoutException e){ //退出 f.cancel(false); }线程设计SpyMemcached有两类线程：业务线程和selector线程业务线程：封装请求task、对象序列化、封装发送的协议、并将task放到对应连接的队列中对收到的数据进行反序列化为对象Selector线程读取连接中的队列，将队列中的task的数据发送到memcached读取Memcached返回的数据，解析协议并通知业务线程处理对失败的节点进行自动重连sharding机制路由机制Spymemcached默认的hash算法有：arrayMod：传统hash 模节点取余hash(key)%节点数，即余数ketama：一致性hash（推荐） 扩容、缩容好些容错key路由到服务节点，服务节点宕机，有两种方式处理：自动重连失败的节点从正常的队列中摘除添加到重连队列中Selector线程定期对该节点进行重连failover处理Redistribute(推荐)遇到失败节点后选择下一个节点，直到选到正常的节点缺点：大量回源Retry继续访问，一般就失败了Cancel抛异常退出访问Memcache的集群处理容错比较差序列化我们知道在日常存储中数据往往是对象，那么在Memcached中会以二进制的方式存储，所以在存数据时要进行对象的序列化（jdk序列化），在取数据时要进行对象的反序列化，这个过程是由业务线程处理序列化后会判断长度是否大于阈值16384个byte，如果大于将采用Gzip的方式进行数据压缩
Guava Cache介绍JVM缓存JVM 缓存，是堆缓存。其实就是创建一些全局容器，比如List、Set、Map等，这些容器用来做数据存储。这样做的问题：不能按照一定的规则淘汰数据，如 LRU，LFU，FIFO 等。没有清除数据时的回调通知并发处理能力差，针对并发可以使用CurrentHashMap，但缓存的其他功能需要自行实现缓存过期处理，缓存数据加载刷新等都需要手工实现Guava CacheGuava是Google提供的一套Java工具包，而Guava Cache是一套非常完善的本地缓存机制（JVM缓存）。Guava cache的设计来源于CurrentHashMap，可以按照多种策略来清理存储在其中的缓存值且保持很高的并发读写性能。Guava Cache应用场景本地缓存的应用场景：对性能有非常高的要求不经常变化占用内存不大有访问整个集合的需求数据允许不时时一致guava cache高并发，不需要持久化currentHashMap高并发Ehcached持久化 二级缓存Guava Cache 的优势：缓存过期和淘汰机制在GuavaCache中可以设置Key的过期时间，包括访问过期和创建过期GuavaCache在缓存容量达到指定大小时，采用LRU的方式，将不常使用的键值从Cache中删除并发处理能力GuavaCache类似CurrentHashMap，是线程安全的。提供了设置并发级别的api，使得缓存支持并发的写入和读取采用分离锁机制，分离锁能够减小锁力度，提升并发能力，分离锁是分拆锁定，把一个集合看分成若干partition，每个partiton一把锁。（同ConcurrentHashMap）就是分了16个区域，这16个区域之间是可以并发的。GuavaCache采用Segment做分区。更新锁定一般情况下，在缓存中查询某个key，如果不存在，则查源数据，并回填缓存。（Cache Aside Pattern）在高并发下会出现，多次查源并重复回填缓存，可能会造成源的宕机（DB），性能下降GuavaCache可以在CacheLoader的load方法中加以控制，对同一个key，只让一个请求去读源并回填缓存，其他请求阻塞等待。集成数据源一般我们在业务中操作缓存，都会操作缓存和数据源两部分而GuavaCache的get可以集成数据源，在从缓存中读取不到时可以从数据源中读取数据并回填缓存监控缓存加载/命中情况统计Guava Cache创建方式GuavaCache有两种创建方式：CacheLoader和Callable callbackpublic class GuavaDemo { public static void main(String args[]) throws Exception { LoadingCache<String,Object> cache = CacheBuilder.newBuilder() // 最大3个 .maximumSize(3) //Cache中存储的对象,写入3秒后过期 .expireAfterWrite(3,TimeUnit.SECONDS) //记录命中率 .recordStats() //失效通知 .removalListener(new RemovalListener<Object, Object>() { public void onRemoval(RemovalNotification<Object, Object> notification){ System.out.println(notification.getKey()+":"+notification.getCause()); } } .build( new CacheLoader<String, Object>() { @Override public String load(String s) throws Exception { return Constants.hm.get(s); } } ); //初始化cache initCache(cache); System.out.println(cache.size()); displayCache(cache); System.out.println("============================="); Thread.sleep(1000); System.out.println(cache.getIfPresent("1")); Thread.sleep(2500); System.out.println("============================="); displayCache(cache); } public static Object get(String key,LoadingCache cache) throws Exception{ Object value=cache.get(key, new Callable() { @Override public Object call() throws Exception { Object v= Constants.hm.get(key); //设置回缓存 cache.put(key,v); return v; } } ); return value; } public static void initCache(LoadingCache cache) throws Exception{ //前三条记录 for (int i=1;i<=3;i++){ cache.get(String.valueOf(i)); } } //获得当前缓存的记录 public static void displayCache(LoadingCache cache) throws Exception{ Iterator its=cache.asMap().entrySet().iterator(); while(its.hasNext()){ System.out.println(its.next().toString()); } } }CacheLoader在创建cache对象时，采用CacheLoader来获取数据，当缓存不存在时能够自动加载数据到缓存中。LoadingCache<String,String> cache = CacheBuilder.newBuilder() .maximumSize(3) .build( new CacheLoader<String, String>() { @Override public String load(String s) throws Exception { return Constants.hm.get(s); } } );Callable Callbackpublic static Object get(String key,LoadingCache cache) throws Exception{ Object value=cache.get(key, new Callable() { @Override public Object call() throws Exception { String v= Constants.hm.get(key); //设置回缓存（这句不写也可以，默认自动设置回缓存） cache.put(key,v); return v; } } ); return value; }缓存数据删除GuavaCache的数据删除分为：被动删除和主动删除被动删除基于数据大小的删除LoadingCache<String,Object> cache= CacheBuilder.newBuilder() // TODO 加附加的功能 //最大个数 .maximumSize(3) .build(new CacheLoader<String, Object>() { //读取数据源 @Override public Object load(String key) throws Exception { return Constants.hm.get(key); } } ); //读取缓存中的1的数据 缓存有就读取 没有就返回null System.out.println(cache.getIfPresent("5")); //读取4 读源并回写缓存 淘汰一个（LRU+FIFO） get("4",cache);规则：LRU+FIFO访问次数一样少的情况下，FIFO基于过期时间的删除隔多长时间后没有被访问过的key被删除//缓存中的数据 如果3秒内没有访问则删除 .maximumSize(3).expireAfterAccess(3, TimeUnit.SECONDS) // TODO 这里省略build Thread.sleep(1000); //访问1 1被访问 cache.getIfPresent("1"); //歇了2.1秒 Thread.sleep(2100); //最后缓存中会留下1 System.out.println("=================================="); display(cache);写入多长时间后过期//等同于expire ttl 缓存中对象的生命周期就是3秒 .maximumSize(3).expireAfterWrite(3, TimeUnit.SECONDS) .build(new CacheLoader<String, Object>() { //读取数据源 @Override public Object load(String key) throws Exception { return Constants.hm.get(key); } } ); display(cache); Thread.sleep(1000); //访问1 cache.getIfPresent("1"); //歇了2.1秒 Thread.sleep(2100); System.out.println("=================================="); display(cache);基于引用的删除可以通过weakKeys和weakValues方法指定Cache只保存对缓存记录key和value的弱引用。这样当没有其他强引用指向key和value时，key和value对象就会被垃圾回收器回收。LoadingCache<String,Object> cache = CacheBuilder.newBuilder() // 最大3个 值的弱引用 .maximumSize(3).weakValues() .build( ); Object value = new Object(); cache.put("1",value); //原对象不再有强引用 value = new Object(); //强制垃圾回收 System.gc(); System.out.println(cache.getIfPresent("1"));主动删除单独删除//将key=1 删除 cache.invalidate("1");批量删除//将key=1和2的删除 cache.invalidateAll(Arrays.asList("1","2"));清空所有数据//清空缓存 cache.invalidateAll();Guava Cache原理GuavaCache核心原理之数据结构Guava Cache的数据结构跟ConcurrentHashMap类似，但也不完全一样。最基本的区别是ConcurrentMap会一直保存所有添加的元素，直到显式地移除。相对地，Guava Cache为了限制内存占用，通常都设定为自动回收元素。其数据结构图如下：LocalCache为Guava Cache的核心类，包含一个Segment数组组成Segement数组的长度决定了cache的并发数每一个Segment使用了单独的锁，其实每个Segment继承了ReentrantLock，对Segment的写操作需要先拿到锁每个Segment由一个table和5个队列组成5个队列：ReferenceQueue keyReferenceQueue ： 已经被GC，需要内部清理的键引用队列ReferenceQueue valueReferenceQueue ： 已经被GC，需要内部清理的值引用队列ConcurrentlinkedQueue<ReferenceEntry<k,v>> recencyQueue : LRU队列，当segment上达到临界值发生写操作时该队列会移除数据Queue<ReferenceEntry<K, V>> writeQueue：写队列，按照写入时间进行排序的元素队列，写入一个元素时会把它加入到队列尾部Queue<ReferenceEntry<K, V>> accessQueue：访问队列，按照访问时间进行排序的元素队列，访问(包括写入)一个元素时会把它加入到队列尾部1个table：AtomicReferenceArray<ReferenceEntry<K, V>> table：AtomicReferenceArray可以用原子方式更新其元素的对象引用数组ReferenceEntry<k,v>ReferenceEntry是Guava Cache中对一个键值对节点的抽象，每个ReferenceEntry数组项都是一条ReferenceEntry链。并且一个ReferenceEntry包含key、hash、valueReference、next字段（单链）Guava Cache使用ReferenceEntry接口来封装一个键值对，而用ValueReference来封装Value值GuavaCache核心原理之回收机制Guava Cache提供了三种基本的缓存回收方式：基于容量回收在缓存项的数目达到限定值之前，采用LRU的回收方式定时回收expireAfterAccess：缓存项在给定时间内没有被读/写访问，则回收。回收顺序和基于大小回收一样（LRU）expireAfterWrite：缓存项在给定时间内没有被写访问（创建或覆盖），则回收基于引用回收通过使用弱引用的键、或弱引用的值、或软引用的值，Guava Cache可以垃圾回收除了以上三种还有主动删除，采用命令，这个前面讲了GuavaCache构建的缓存不会“自动”执行清理和回收工作，也不会在某个缓存项过期后马上清理，也没有诸如此类的清理机制。GuavaCache是在每次进行缓存操作的时候，惰性删除 如get()或者put()的时候，判断缓存是否过期GuavaCache核心原理之Segment定位先通过key做hash定位到所在的Segment通过位运算找首地址的偏移量 SegmentCount>=并发数且为2的n次方V get(K key, CacheLoader<? super K, V> loader) throws ExecutionException { // 注意，key不可为空 int hash = hash(checkNotNull(key)); // 通过hash定位到segment数组的某个Segment元素，然后调用其get方法 return segmentFor(hash).get(key, hash, loader); }再找到segment中的Entry链数组，通过key的hash定位到某个Entry节点V get(K key, int hash, CacheLoader<? super K, V> loader) throws ExecutionException { checkNotNull(key); checkNotNull(loader); try { if (count != 0) { // read-volatile // 内部也是通过找Entry链数组定位到某个Entry节点 ReferenceEntry<K, V> e = getEntry(key, hash); ......Guava Cache高级实战GuavaCache高级实战之并发操作并发设置GuavaCache通过设置 concurrencyLevel 使得缓存支持并发的写入和读取LoadingCache<String,Object> cache = CacheBuilder.newBuilder() // 最大3个 同时支持CPU核数线程写缓存 .maximumSize(3).concurrencyLevel(Runtime.getRuntime().availableProcessors()).build();concurrencyLevel=Segment数组的长度同ConcurrentHashMap类似Guava cache的并发也是通过分离锁实现V get(K key, CacheLoader<? super K, V> loader) throws ExecutionException { int hash = this.hash(Preconditions.checkNotNull(key)); //通过hash值确定该key位于哪一个segment上，并获取该segment return this.segmentFor(hash).get(key, hash, loader); }LoadingCache采用了类似ConcurrentHashMap的方式，将映射表分为多个segment。segment之间可以并发访问，这样可以大大提高并发的效率，使得并发冲突的可能性降低了。更新锁定GuavaCache提供了一个refreshAfterWrite定时刷新数据的配置项如果经过一定时间没有更新或覆盖，则会在下一次获取该值的时候，会在后台异步去刷新缓存刷新时只有一个请求回源取数据，其他请求会阻塞（block）在一个固定时间段，如果在该时间段内没有获得新值则返回旧值。LoadingCache<String,Object> cache = CacheBuilder.newBuilder() // 最大3个 同时支持CPU核数线程写缓存 .maximumSize(3).concurrencyLevel(Runtime.getRuntime().availableProcessors()) //3秒内阻塞会返回旧数据 .refreshAfterWrite(3,TimeUnit.SECONDS).build();案例分享商城首页：类目public class Bean { private int id; private String name; private int pid; private String type; private String url; private int isHot; }更新锁定：accesstoken token失效，从公网拿token，采用更新锁定GuavaCache高级实战之动态加载动态加载行为发生在获取不到数据或者是数据已经过期的时间点，Guava动态加载使用回调模式用户自定义加载方式，然后Guava cache在需要加载新数据时会回调用户的自定义加载方式segmentFor(hash).get(key, hash, loader)loader即为用户自定义的数据加载方式，当某一线程get不到数据会去回调该自定义加载方式去加载数据GuavaCache高级实战之自定义LRU算法public class LinkedHashLRUcache<k, v> { /** * LinkedHashMap（自身实现了ＬＲＵ算法） * 有序 * 每次访问一个元素，都会加到尾部 */ int limit; LRUcache<k, v> internalLRUcache; public LinkedHashLRUcache(int limit) { this.limit = limit; this.internalLRUcache = new LRUcache(limit); } public void put(k key, v value) { this.internalLRUcache.put(key, value); } public v get(k key) { return this.internalLRUcache.get(key); } public static void main(String[] args) { LinkedHashLRUcache lru=new LinkedHashLRUcache(3); lru.put(1,"zhangfei1"); lru.put(2,"zhangfei2"); lru.put(3,"zhangfei3"); lru.get(1); lru.put(4,"zhangfei4"); for (Object o:lru.internalLRUcache.values()){ System.out.println(o.toString()); } } } public class LRUcache<k, v> extends LinkedHashMap<k, v> { private final int limit; public LRUcache(int limit) { //初始化 accessOrder ： true 改变尾结点 super(16, 0.75f, true); this.limit = limit; } //是否删除最老的数据 @Override protected Boolean removeEldestEntry(Map.Entry<k, v> eldest) { return size() > limit; } }GuavaCache高级实战之疑难问题GuavaCache会oom（内存溢出）吗会，当我们设置缓存永不过期（或者很长），缓存的对象不限个数（或者很大）时，比如：Cache<String, String> cache = CacheBuilder.newBuilder() .expireAfterWrite(100000, TimeUnit.SECONDS) .build();不断向GuavaCache加入大字符串，最终将会oom解决方案：缓存时间设置相对小些，使用弱引用方式存储对象Cache<String, String> cache = CacheBuilder.newBuilder() .expireAfterWrite(1, TimeUnit.SECONDS) .weakValues().build();GuavaCache缓存到期就会立即清除吗不是的，GuavaCache是在每次进行缓存操作的时候，如get()或者put()的时候，判断缓存是否过期void evictEntries(ReferenceEntry<K, V> e) { drainRecencyQueue(); while ((e = writeQueue.peek()) != null && map.isExpired(e, now)) { if (!removeEntry(e, e.getHash(), RemovalCause.EXPIRED)) { throw new AssertionError(); } } while ((e = accessQueue.peek()) != null && map.isExpired(e, now)) { if (!removeEntry(e, e.getHash(), RemovalCause.EXPIRED)) { throw new AssertionError(); } } }一个如果一个对象放入缓存以后，不在有任何缓存操作（包括对缓存其他key的操作），那么该缓存不会主动过期的。GuavaCache如何找出最久未使用的数据用accessQueue，这个队列是按照LRU的顺序存放的缓存对象（ReferenceEntry）的。会把访问过的对象放到队列的最后。并且可以很方便的更新和删除链表中的节点，因为每次访问的时候都可能需要更新该链表，放入到链表的尾部。这样，每次从access中拿出的头节点就是最久未使用的。对应的writeQueue用来保存最久未更新的缓存队列，实现方式和accessQueue一样。Guava Cache源码剖析GuavaCache源码剖析之实现框架GuavaCache体系类图：CacheBuilder：类，缓存构建器。构建缓存的入口，指定缓存配置参数并初始化本地缓存。CacheBuilder在build方法中，会把前面设置的参数，全部传递给LocalCache，它自己实际不参与任何计算CacheLoader：抽象类。用于从数据源加载数据，定义load、reload、loadAll等操作Cache：接口，定义get、put、invalidate等操作，这里只有缓存增删改的操作，没有数据加载的操作LoadingCache：接口，继承自Cache。定义get、getUnchecked、getAll等操作，这些操作都会从数据源load数据LocalCache：类。整个guava cache的核心类，包含了guava cache的数据结构以及基本的缓存的操作方法LocalManualCache：LocalCache内部静态类，实现Cache接口。其内部的增删改缓存操作全部调用成员变量localCache（LocalCache类型）的相应方法LocalLoadingCache：LocalCache内部静态类，继承自LocalManualCache类，实现LoadingCache接口。其所有操作也是调用成员变量localCache（LocalCache类型）的相应方法LocalCacheLoadingCache这些类表示获取Cache的方式，可以有多种方式，但是它们的方法最终调用到LocalCache的方法，LocalCache是Guava Cache的核心类。class LocalCache<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>LocalCache为Guava Cache的核心类 LocalCache的数据结构与ConcurrentHashMap很相似，都由多个segment组成，且各segment相对独立，互不影响，所以能支持并行操作。//Map的数组 final Segment<K, V>[] segments; //并发量，即segments数组的大小 final int concurrencyLevel; ... //访问后的过期时间，设置了expireAfterAccess就有 final long expireAfterAccessNanos; //写入后的过期时间，设置了expireAfterWrite就有 final long expireAfterWriteNa就有nos; //刷新时间，设置了refreshAfterWrite就有 final long refreshNanos; //removal的事件队列，缓存过期后先放到该队列 final Queue<RemovalNotification<K, V>> removalNotificationQueue; //设置的removalListener final RemovalListener<K, V> removalListener; ...每个segment由一个table和若干队列组成。缓存数据存储在table中，其类型为AtomicReferenceArray。static class Segment<K, V> extends ReentrantLock{ /** * segments 维护一个entry列表的table，确保一致性状态。所以可以不加锁去读。节点的next field是不可修改的final，因为所有list的增加操作 */ final LocalCache<K, V> map; /** * 该segment区域内所有存活的元素个数 */ volatile int count; /** * 改变table大小size的更新次数。这个在批量读取方法期间保证它们可以看到一致性的快照： * 如果modCount在我们遍历段加载大小或者核对containsValue期间被改变了，然后我们会看到一个不一致的状态视图，以至于必须去重试。 * count+modCount 保证内存一致性 * * 感觉这里有点像是版本控制，比如数据库里的version字段来控制数据一致性 */ int modCount; /** * 每个段表，使用乐观锁的Array来保存entry The per-segment table. */ // 这里和concurrentHashMap不一致，原因是这边元素是引用，直接使用不会线程安全 volatile AtomicReferenceArray<ReferenceEntry<K, V>> table; /** * A queue of elements currently in the map, ordered by write time.Elements are added to the tail of the queue * on write. */ @GuardedBy("Segment.this") final Queue<ReferenceEntry<K, V>> writeQueue; /** * A queue of elements currently in the map, ordered by access time.Elements are added to the tail of the queue on access (note that writes count as accesses). */ @GuardedBy("Segment.this") final Queue<ReferenceEntry<K, V>> accessQueue; } interface ReferenceEntry<K, V> { /** * Returns the value reference from this entry. */ ValueReference<K, V> getValueReference(); /** * Sets the value reference for this entry. */ void setValueReference(ValueReference<K, V> valueReference); /** * Returns the next entry in the chain. */ @Nullable ReferenceEntry<K, V> getNext(); /** * Returns the entry's hash. */ int getHash(); /** * Returns the key for this entry. */ @Nullable K getKey(); /* * Used by entries that use access order. Access entries are maintained in a doubly-linked list. * New entries are added at the tail of the list at write time; stale entries are expired from * the head of the list. */ /** * Returns the time that this entry was last accessed, in ns. */ long getAccessTime(); /** * Sets the entry access time in ns. */ void setAccessTime(long time); }GuavaCache源码剖析之CacheBuilder缓存构建器。构建缓存的入口，指定缓存配置参数并初始化本地缓存。主要采用builder的模式，CacheBuilder的每一个方法都返回这个CacheBuilder知道build方法的调用。注意build方法有重载，带有参数的为构建一个具有数据加载功能的缓存，不带参数的构建一个没有数据加载功能的缓存。LocalLoadingCache( CacheBuilder<? super K, ? super V> builder, CacheLoader<? super K, V> loader) { //LocalLoadingCache构造函数需要一个LocalCache作为参数 super(new LocalCache<K, V>(builder, checkNotNull(loader))); } //构造LocalCache LocalCache( CacheBuilder<? super K, ? super V> builder, @Nullable CacheLoader<? super K, V> loader) { //默认并发水平是4 concurrencyLevel = Math.min(builder.getConcurrencyLevel(), MAX_SEGMENTS); //key的强引用 keyStrength = builder.getKeyStrength(); valueStrength = builder.getValueStrength(); //key比较器 keyEquivalence = builder.getKeyEquivalence(); valueEquivalence = builder.getValueEquivalence(); maxWeight = builder.getMaximumWeight(); weigher = builder.getWeigher(); //读写后有效期，超时重载 expireAfterAccessNanos = builder.getExpireAfterAccessNanos(); //写后有效期，超时重载 expireAfterWriteNanos = builder.getExpireAfterWriteNanos(); refreshNanos = builder.getRefreshNanos(); //缓存触发失效 或者 GC回收软/弱引用，触发监听器 removalListener = builder.getRemovalListener(); //移除通知队列 removalNotificationQueue = (removalListener == NullListener.INSTANCE) ? LocalCache.<RemovalNotification<K, V>>discardingQueue() : new ConcurrentLinkedQueue<RemovalNotification<K, V>>(); ticker = builder.getTicker(recordsTime()); entryFactory = EntryFactory.getFactory(keyStrength, usesAccessEntries(), usesWriteEntries()); globalStatsCounter = builder.getStatsCounterSupplier().get(); //缓存加载器 defaultLoader = loader; int initialCapacity = Math.min(builder.getInitialCapacity(), MAXIMUM_CAPACITY); if (evictsBySize() && !customWeigher()) { initialCapacity = Math.min(initialCapacity, (int) maxWeight); }GuavaCache源码剖析之Put流程上锁清除队列元素清理的是keyReferenceQueue和valueReferenceQueue这两个队列，这两个队列是引用队列如果发现key或者value被GC了，那么会在put的时候触发清理setvalue方法了，它做的是将value写入EntryV put(K key, int hash, V value, Boolean onlyIfAbsent) { //保证线程安全，加锁 lock(); try { //获取当前的时间 long now = map.ticker.read(); //清除队列中的元素 preWriteCleanup(now); ... //获取当前Entry中的HashTable的Entry数组 AtomicReferenceArray<ReferenceEntry<K, V>> table = this.table; //定位 int index = hash & (table.length() - 1); //获取第一个元素 ReferenceEntry<K, V> first = table.get(index); //遍历整个Entry链表 // Look for an existing entry. for (ReferenceEntry<K, V> e = first; e != null; e = e.getNext()) { K entryKey = e.getKey(); if (e.getHash() == hash && entryKey != null && map.keyEquivalence.equivalent(key, entryKey)) { // We found an existing entry. //如果找到相应的元素 ValueReference<K, V> valueReference = e.getValueReference(); //获取value V entryValue = valueReference.get(); //如果entry的value为null，可能被GC掉了 if (entryValue == null) { ++modCount; if (valueReference.isActive()) { //减小锁时间的开销 enqueueNotification( key, hash, entryValue, valueReference.getWeight(), RemovalCause.COLLECTED); //利用原来的key并且刷新value //存储数据，并且将新增加的元素写入两个队列中，一个Write队列、一个Access队列 setValue(e, key, value, now); // count remains unchanged newCount = this.count; } else { //存储数据，并且将新增加的元素写入两个队列中 setValue(e, key, value, now); newCount = this.count + 1; } this.count = newCount; // write-volatile，保证内存可见性 //淘汰缓存 evictEntries(e); return null; //原来的Entry中包含指定key的元素，所以读取一次，读取操作需要更新Access队列 } else if (onlyIfAbsent) { ....... //存储数据，并且将新增加的元素写入两个队列中 setValue(e, key, value, now); //数据的淘汰 evictEntries(e); return entryValue; } } } //如果目标的entry不存在，那么新建entry // Create a new entry. ++modCount; ReferenceEntry<K, V> newEntry = newEntry(key, hash, first); setValue(newEntry, key, value, now); ...... } finally { //解锁 unlock(); //处理刚刚的remove Cause postWriteCleanup(); } }GuavaCache源码剖析之Get流程获取对象引用（引用可能是非alive的，比如是需要失效的、比如是loading的）；判断对象引用是否是alive的（如果entry是非法的、部分回收的、loading状态、需要失效的，则认为不是alive）。如果对象是alive的，如果设置refresh，则异步刷新查询value，然后等待返回最新value。针对不是alive的，但却是在loading的，等待loading完成（阻塞等待）。这里如果value还没有拿到，则查询loader方法获取对应的值（阻塞获取）。// LoadingCache methods //local cache的代理 V get(K key, CacheLoader<? super K, V> loader) throws ExecutionException { //hash——>rehash int hash = hash(checkNotNull(key)); return segmentFor(hash).get(key, hash, loader); } // loading //进行指定key对应的value的获取，读取不加锁 V get(K key, int hash, CacheLoader<? super K, V> loader) throws ExecutionException { .... try { // read-volatile volatile读会刷新缓存，尽量保证可见性,如果为0那么直接load if (count != 0) { // don't call getLiveEntry, which would ignore loading values ReferenceEntry<K, V> e = getEntry(key, hash); //如果对应的Entry不为Null,证明值还在 if (e != null) { //获取当前的时间，根据当前的时间进行Live的数据的读取 long now = map.ticker.read(); // 判断是否为alive（此处是懒失效，在每次get时才检查是否达到失效时机） V value = getLiveValue(e, now); ...... } } //如果取不到值，那么进行统一的加锁get // at this point e is either null or expired; 此处或者为null，或者已经被失效。 return lockedGetOrLoad(key, hash, loader); } catch (ExecutionException ee) { Throwable cause = ee.getCause(); if (cause instanceof Error) { throw new ExecutionError((Error) cause); } else if (cause instanceof RuntimeException) { throw new UncheckedExecutionException(cause); } throw ee; } finally { //每次Put和get之后都要进行一次Clean postReadCleanup(); } }GuavaCache源码剖析之过期重载数据过期不会自动重载，而是通过get操作时执行过期重载。具体就是CacheBuilder构造的LocalLoadingCachestatic class LocalLoadingCache<K, V> extends LocalManualCache<K, V> implements LoadingCache<K, V> { LocalLoadingCache(CacheBuilder<? super K, ? super V> builder, CacheLoader<? super K, V> loader) { super(new LocalCache<K, V>(builder, checkNotNull(loader))); } // LoadingCache methods @Override public V get(K key) throws ExecutionException { return localCache.getOrLoad(key); } @Override public V getUnchecked(K key) { try { return get(key); } catch (ExecutionException e) { throw new UncheckedExecutionException(e.getCause()); } } @Override public ImmutableMap<K, V> getAll(Iterable<? extends K> keys) throws ExecutionException { return localCache.getAll(keys); } @Override public void refresh(K key) { localCache.refresh(key); } @Override public final V apply(K key) { return getUnchecked(key); } // Serialization Support private static final long serialVersionUID = 1; @Override Object writeReplace() { return new LoadingSerializationProxy<K, V>(localCache); } }
使用场景：互联网，数据越来越多，用户越来越多，并发量、吞吐量越来越大使用数据库存储，分库分表，也不能满足要求，使用缓存，减轻数据库的压力临时存储的数据其他的场景：Session分离、分布式锁缓存（服务器端）：本地缓存（local cache）CurrentHashMap、Ehcache、Guava Cache缓存在应用服务器，全局变量，JVM缓存优势：和应用程序在同一进程，访问快，没有网络开销，一般不会崩缺点：容量小，每个JVM有一份，有数据冗余分布式缓存Redis、Memcached、Tair（阿里、美团）、EVCache（AWS）、Aerospike优势：空间优势、在应用中共享（session共享）、高可用（主从）、高扩展（分区）、集群缺点：资源、网络开销、数据一致性（AP）前面已经详解了Redis，接下来讲其他的缓存GuavaCache、Tair、EVCache、Aerospike比较 GuavaCacheTairEVCacheAerospike类别本地缓存（JVM缓存）分布式缓存分布式缓存分布式NoSQL数据库应用高并发本地缓存阿里、美团Netflix、AWS互联网广告行业（国外）性能高较高很高较高持久化无有有有集群无有有有