TrumanDu's Blog

//不指定前缀和后缀
StringJoiner stringJoiner = new StringJoiner(",");
//指定前缀和后缀
//StringJoiner stringJoiner = new StringJoiner(",","{","}");
List<String> list = Arrays.asList("a","b","c");
list.forEach(str->stringJoiner.add(str));

List<String> list = Arrays.asList("a","b","c");
System.out.println(String.join(",", list));

Clock clock = Clock.systemDefaultZone();
long millis = clock.millis();

Instant instant = clock.instant();
Date legacyDate = Date.from(instant);   // legacy java.util.Date

System.out.println(ZoneId.getAvailableZoneIds());
// prints all available timezone ids

ZoneId shanghaiZoneId = ZoneId.of("Asia/Shanghai");
// ZoneRules[currentStandardOffset=+08:00]

LocalDateTime localDateTime = LocalDateTime.now();
ZonedDateTime zonedDateTime = ZonedDateTime.of(localDateTime, shanghaiZoneId);

LocalTime now1 = LocalTime.now(zone1);
LocalTime now2 = LocalTime.now(zone2);

System.out.println(now1.isBefore(now2));  // false

long hoursBetween = ChronoUnit.HOURS.between(now1, now2);
long minutesBetween = ChronoUnit.MINUTES.between(now1, now2);

System.out.println(hoursBetween);       // -3
System.out.println(minutesBetween);     // -239

LocalDate today = LocalDate.now();
LocalDate tomorrow = today.plus(1, ChronoUnit.DAYS);
LocalDate yesterday = tomorrow.minusDays(2);

LocalDate independenceDay = LocalDate.of(2014, Month.JULY, 4);
DayOfWeek dayOfWeek = independenceDay.getDayOfWeek();
System.out.println(dayOfWeek);    // FRIDAY

LocalDateTime sylvester = LocalDateTime.of(2014, Month.DECEMBER, 31, 23, 59, 59);

DayOfWeek dayOfWeek = sylvester.getDayOfWeek();
System.out.println(dayOfWeek);      // WEDNESDAY

Month month = sylvester.getMonth();
System.out.println(month);          // DECEMBER

long minuteOfDay = sylvester.getLong(ChronoField.MINUTE_OF_DAY);
System.out.println(minuteOfDay);    // 1439

ZonedDateTime dateTime = ZonedDateTime.now();
ZonedDateTime zonedDateTime = ZonedDateTime.of(LocalDateTime.now(), shanghaiZoneId);

Instant first = Instant.now();
// wait some time while something happens
Instant second = Instant.now();
Duration duration = Duration.between(first, second);

BASIC_ISO_DATE

ISO_LOCAL_DATE
ISO_LOCAL_TIME
ISO_LOCAL_DATE_TIME

ISO_OFFSET_DATE
ISO_OFFSET_TIME
ISO_OFFSET_DATE_TIME

ISO_ZONED_DATE_TIME

ISO_INSTANT

ISO_DATE
ISO_TIME
ISO_DATE_TIME

ISO_ORDINAL_TIME
ISO_WEEK_DATE

RFC_1123_DATE_TIME

DateTimeFormatter formatter = DateTimeFormatter.ISO_DATE_TIME;
//DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String formattedDate = formatter.format(LocalDateTime.now());

$ bin/kafka-consumer-groups.sh --describe --bootstrap-server 192.168.0.101:8092 --group test
Consumer group 'test' has no active members.

TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
truman_test_offset 2          1325            2361            1036            -               -               -
truman_test_offset 6          1265            2289            1024            -               -               -
truman_test_offset 4          1245            2243            998             -               -               -
truman_test_offset 9          1310            2307            997             -               -               -
truman_test_offset 1          1259            2257            998             -               -               -
truman_test_offset 8          1410            2438            1028            -               -               -
truman_test_offset 3          1225            2167            942             -               -               -
truman_test_offset 0          1218            2192            974             -               -               -
truman_test_offset 5          1262            2252            990             -               -               -
truman_test_offset 7          1265            2277            1012            -               -               -

public class TopicConsumerGroupState {
	private String groupId;
	/**
	 * 消费方式：zk/broker 主要指的offset提交到哪里 新版本 broker 旧版本zk
	 */
	private String consumerMethod;
	private List<PartitionAssignmentState> partitionAssignmentStates;
	/**
	 * Dead：组内已经没有任何成员的最终状态，组的元数据也已经被coordinator移除了。这种状态响应各种请求都是一个response：
	 * UNKNOWN_MEMBER_ID Empty：组内无成员，但是位移信息还没有过期。这种状态只能响应JoinGroup请求
	 * PreparingRebalance：组准备开启新的rebalance，等待成员加入 AwaitingSync：正在等待leader
	 * consumer将分配方案传给各个成员 Stable：rebalance完成！可以开始消费了~
	 */
	private ConsumerGroupState consumerGroupState;

    //省略set和get方法..

}

public class PartitionAssignmentState {
	private String group; 
	private String topic;
	private int partition;
	private long offset;
	private long lag;
	private String consumerId;
	private String host;
	private String clientId;
	private long logEndOffset;
	//省略set和get方法..
}

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.2.0</version>
</dependency>

/**
	 * 根据topic 获取offset,该结果是不同group 组不同patition的当前消费offset
	 * 
	 * @param topic
	 * @return Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>>
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>> listConsumerGroupOffsets(String topic)
			throws InterruptedException, ExecutionException, TimeoutException {
		Set<String> groupIds = this.listConsumerGroups(topic);
		Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>> consumerGroupOffsets = new HashMap<>();
		groupIds.forEach(groupId -> {
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets = new HashSet<>();
			try {
				consumerPatitionOffsets = this.adminClient.listConsumerGroupOffsets(groupId)
						.partitionsToOffsetAndMetadata().get(30, TimeUnit.SECONDS).entrySet().stream()
						.filter(entry -> topic.equalsIgnoreCase(entry.getKey().topic())).collect(Collectors.toSet());
				;
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			} catch (TimeoutException e) {
				e.printStackTrace();
			}
			consumerGroupOffsets.put(groupId, consumerPatitionOffsets);
		});
		return consumerGroupOffsets;
	}

	/**
	 * 根据topic 获取topicConsumerGroupStates,其中未包含lag/logEndOffset(consumer可见offset)
	 * 
	 * @param topic
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public List<TopicConsumerGroupState> describeConsumerGroups(String topic)
			throws InterruptedException, ExecutionException, TimeoutException {
		final List<TopicConsumerGroupState> topicConsumerGroupStates = new ArrayList<>();
		Set<String> groupIds = this.listConsumerGroups(topic);
		Map<String, ConsumerGroupDescription> groupDetails = this.adminClient.describeConsumerGroups(groupIds).all()
				.get(30, TimeUnit.SECONDS);

		Map<String, Set<Entry<TopicPartition, OffsetAndMetadata>>> consumerPatitionOffsetMap = this
				.listConsumerGroupOffsets(topic);

		groupDetails.entrySet().forEach(entry -> {
			String groupId = entry.getKey();
			ConsumerGroupDescription description = entry.getValue();

			TopicConsumerGroupState topicConsumerGroupState = new TopicConsumerGroupState();
			topicConsumerGroupState.setGroupId(groupId);
			topicConsumerGroupState.setConsumerMethod("broker");
			topicConsumerGroupState.setConsumerGroupState(description.state());
			// 获取group下不同patition消费offset信息
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets = consumerPatitionOffsetMap
					.get(groupId);
			List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();

			if (!description.members().isEmpty()) {
				// 获取存在consumer(memeber存在的情况)
				partitionAssignmentStates = this.withMembers(consumerPatitionOffsets, topic, groupId, description);
			} else {
				// 获取不存在consumer
				partitionAssignmentStates = this.withNoMembers(consumerPatitionOffsets, topic, groupId);
			}
			topicConsumerGroupState.setPartitionAssignmentStates(partitionAssignmentStates);
			topicConsumerGroupStates.add(topicConsumerGroupState);
		});

		return topicConsumerGroupStates;
	}

	private List<PartitionAssignmentState> withMembers(
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets, String topic, String groupId,
			ConsumerGroupDescription description) {
		List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();
		Map<Integer, Long> consumerPatitionOffsetMap = new HashMap<>();
		consumerPatitionOffsets.forEach(entryInfo -> {
			TopicPartition topicPartition = entryInfo.getKey();
			OffsetAndMetadata offsetAndMetadata = entryInfo.getValue();
			consumerPatitionOffsetMap.put(topicPartition.partition(), offsetAndMetadata.offset());
		});
		description.members().forEach(memberDescription -> {
			memberDescription.assignment().topicPartitions().forEach(topicPation -> {
				PartitionAssignmentState partitionAssignmentState = new PartitionAssignmentState();
				partitionAssignmentState.setPartition(topicPation.partition());
				partitionAssignmentState.setTopic(topic);
				partitionAssignmentState.setClientId(memberDescription.clientId());
				partitionAssignmentState.setGroup(groupId);
				partitionAssignmentState.setConsumerId(memberDescription.consumerId());
				partitionAssignmentState.setHost(memberDescription.host());
				partitionAssignmentState.setOffset(consumerPatitionOffsetMap.get(topicPation.partition()));
				partitionAssignmentStates.add(partitionAssignmentState);
			});
		});
		return partitionAssignmentStates;
	}

	private List<PartitionAssignmentState> withNoMembers(
			Set<Entry<TopicPartition, OffsetAndMetadata>> consumerPatitionOffsets, String topic, String groupId) {
		List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();
		consumerPatitionOffsets.forEach(entryInfo -> {
			TopicPartition topicPartition = entryInfo.getKey();
			OffsetAndMetadata offsetAndMetadata = entryInfo.getValue();
			PartitionAssignmentState partitionAssignmentState = new PartitionAssignmentState();
			partitionAssignmentState.setPartition(topicPartition.partition());
			partitionAssignmentState.setTopic(topic);
			partitionAssignmentState.setGroup(groupId);
			partitionAssignmentState.setOffset(offsetAndMetadata.offset());
			partitionAssignmentStates.add(partitionAssignmentState);
		});
		return partitionAssignmentStates;
	}

<dependency>
	<groupId>com.101tec</groupId>
	<artifactId>zkclient</artifactId>
	<version>0.11</version>
</dependency>
<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.11</artifactId>
</dependency>

private List<TopicConsumerGroupState> getZKConsumerOffsets(String topic) {
	final List<TopicConsumerGroupState> topicConsumerGroupStates = new ArrayList<>();
	Set<String> zkGroups = this.listTopicGroups(topic);
	zkGroups.forEach(group -> {
		Map<String, Map<String, String>> zkConsumerOffsets = this.getZKConsumerOffsets(group,
				topic);

		Map<TopicPartition, Long> offsetsMap = this.consumerOffset(group, topic);

		TopicConsumerGroupState topicConsumerGroupState = new TopicConsumerGroupState();
		topicConsumerGroupState.setGroupId(group);
		topicConsumerGroupState.setConsumerMethod("zk");

		List<PartitionAssignmentState> partitionAssignmentStates = new ArrayList<>();

		Map<String, Long> offsetsTempMap = new HashMap<>(offsetsMap.size());
		offsetsMap.forEach((k, v) -> {
			offsetsTempMap.put(k.partition() + "", v);
		});

		zkConsumerOffsets.forEach((patition, topicDesribe) -> {
			Long logEndOffset = offsetsTempMap.get(patition);
			String owner = topicDesribe.get("owner");
			long offset = Long.parseLong(topicDesribe.get("offset"));
			PartitionAssignmentState partitionAssignmentState = new PartitionAssignmentState();
			partitionAssignmentState.setClientId(owner);
			partitionAssignmentState.setGroup(group);
			partitionAssignmentState.setLogEndOffset(logEndOffset);
			partitionAssignmentState.setTopic(topic);
			partitionAssignmentState.setOffset(offset);
			partitionAssignmentState.setPartition(Integer.parseInt(patition));
			partitionAssignmentState.setLag(getLag(offset, logEndOffset));

			partitionAssignmentStates.add(partitionAssignmentState);
		});
		topicConsumerGroupState.setPartitionAssignmentStates(partitionAssignmentStates);
		topicConsumerGroupStates.add(topicConsumerGroupState);
	});
	return topicConsumerGroupStates;
}

npm install antd --save
npm install ant-design-pro@latest --save

// use localStorage to store the authority info, which might be sent from server in actual project.
export function getAuthority() {
  return localStorage.getItem('ice-pro-authority') || 'admin';
}

export function setAuthority(authority) {
  return localStorage.setItem('ice-pro-authority', authority);
}

export function removeAuthority() {
  return localStorage.removeItem('ice-pro-authority');
}

import RenderAuthorized from 'ant-design-pro/lib/Authorized';
import { getAuthority } from './authority';

let Authorized = RenderAuthorized(getAuthority()); // eslint-disable-line

// 更新权限
const reloadAuthorized = () => {
  Authorized = RenderAuthorized(getAuthority());
};

export { reloadAuthorized };
export default Authorized;

this.setState({
      switchStatus: !this.state.switchStatus
    });

<DetailDialog dialogObj={dialogObj} } />

constructor(props) {
      super(props);
      this.state = {
        dialogObj: this.props.dialogObj,
      };
    }

// 接受父类props改变，修改子类中的属性
componentWillReceiveProps(nextProps) {
  this.setState({
    dialogObj: nextProps.dialogObj,
  });
}

<SimpleFormDialog ref="getDialog">

父组件方法
parentMethod = () => {
            //调用组件进行通信
            this.refs.getDialog.childMethod();
}

<DetailDialog hideDetailDialog={this.hideDetailDialog} />

hideDetailDialog = () => {
      this.props.hideDetailDialog();
    }

CREATE STREAM orders (
  orderId BIGINT,
  address STRUCT<street VARCHAR, zip INTEGER>) WITH (...);
  
SELECT address->city, address->zip FROM orders;

sudo docker run -it --rm  confluentinc/cp-ksql-cli:5.1.2  http://192.168.0.11:8088
[sudo] password for bigdata:

                  ===========================================
                  =        _  __ _____  ____  _             =
                  =       | |/ // ____|/ __ \| |            =
                  =       | ' /| (___ | |  | | |            =
                  =       |  <  \___ \| |  | | |            =
                  =       | . \ ____) | |__| | |____        =
                  =       |_|\_\_____/ \___\_\______|       =
                  =                                         =
                  =  Streaming SQL Engine for Apache Kafka® =
                  ===========================================

Copyright 2017-2018 Confluent Inc.

CLI v5.1.2, Server v5.1.2 located at http://192.168.0.11:8088

Having trouble? Type 'help' (case-insensitive) for a rundown of how things work!

ksql> show streams;

 Stream Name      | Kafka Topic              | Format
------------------------------------------------------
 EC_BOT_DETECTION | EC_bot_detection.replica | JSON
 DEMO             | trumantest               | JSON
------------------------------------------------------

curl --request POST \
  --url http://192.168.0.11:8088/ksql \
  --header 'Content-Type: application/vnd.ksql.v1+json; charset=utf-8' \
  --data '{\r\n  "ksql": "LIST STREAMS;",\r\n  "streamsProperties": {}\r\n}'

CREATE STREAM ec_bot_detection (action VARCHAR, nvtc VARCHAR,nid VARCHAR,ip VARCHAR,msg VARCHAR,level VARCHAR,verb VARCHAR,url VARCHAR,purl VARCHAR,ua VARCHAR,valid INTEGER,nNvtc VARCHAR,nNid VARCHAR,cip VARCHAR,sip VARCHAR,xpost VARCHAR,time VARCHAR,id VARCHAR,xprotocol VARCHAR,xhost VARCHAR,xport VARCHAR,xpath VARCHAR,xquery VARCHAR,xpathalias VARCHAR,xkey VARCHAR,importance INTEGER,xpprotocol VARCHAR,xphost VARCHAR,xpport VARCHAR,xppath VARCHAR,xpquery VARCHAR,org VARCHAR,cyc VARCHAR,isp VARCHAR,cycName VARCHAR,region VARCHAR,city VARCHAR,geoLocation VARCHAR,zipcode VARCHAR) WITH (VALUE_FORMAT = 'JSON',KAFKA_TOPIC = 'EC_bot_detection.replica');

SELECT ip,org,cyc,count(1) as ccount FROM EC_BOT_DETECTION WINDOW TUMBLING (SIZE 60 SECONDS) WHERE xhost not like '%.tw' and level='-1' AND importance=1 AND (msg<>'W' OR msg is null) and action like'crawler%' AND ip NOT LIKE '127.0.0.1' AND ip NOT LIKE '172.16.%' AND ip <> ''  AND xpathalias !='LandingpageOverviewcontent4mobile'  AND xpathalias !='CommonCommonrecaptchavalidate'  GROUP BY ip,org,cyc HAVING count(1)>300;

CREATE TABLE bot_detection_1min as SELECT ip,org,cyc,count(1) as ccount FROM EC_BOT_DETECTION WINDOW TUMBLING (SIZE 60 SECONDS) WHERE xhost not like '%.tw' and level='-1' AND importance=1 AND (msg<>'W' OR msg is null) and action like'crawler%' AND ip NOT LIKE '127.0.0.1' AND ip NOT LIKE '172.16.%' AND ip <> ''  AND xpathalias !='LandingpageOverviewcontent4mobile'  AND xpathalias !='CommonCommonrecaptchavalidate'  GROUP BY ip,org,cyc HAVING count(1)>300;

@UdfDescription(name = "upper", description = "字符串转大写")
public class Upper {
	@Udf
	public String upper(final String column) {
		return column.toUpperCase();
	}

}

@UdafDescription(name = "percent", description = "聚合求百分比")
public class Percent {
	@UdafFactory(description = "statistics percent")
	public static Udaf<Integer, Map<String, Double>> createStddev() {
		return new Udaf<Integer, Map<String, Double>>() {
			@Override
			public Map<String, Double> initialize() {
				Map<String, Double> stats = new HashMap<>();
				stats.put("sum", 0.0);
				stats.put("bussiness", 0.0);
				stats.put("percent", 0.0);
				return stats;

			}

			@Override
			public Map<String, Double> aggregate(final Integer val, final Map<String, Double> aggregateValue) {
				Double sum = aggregateValue.getOrDefault("sum", 0.0) + 1;
				Double bussiness = aggregateValue.getOrDefault("bussiness", 0.0);
				if (val % 4 == 0) {
					bussiness = bussiness + 1;
				}

				Double percent = bussiness / sum;

				Map<String, Double> agg = new HashMap<>();
				agg.put("sum", sum);
				agg.put("bussiness", bussiness);
				agg.put("percent", percent);
				return agg;
			}

			@Override
			public Map<String, Double> merge(final Map<String, Double> aggOne, final Map<String, Double> aggTwo) {

				Double sum = aggOne.getOrDefault("sum", 0.0) + aggTwo.getOrDefault("sum", 0.0);
				Double bussiness = aggOne.getOrDefault("bussiness", 0.0) + aggTwo.getOrDefault("bussiness", 0.0);
				Double percent = bussiness / sum;

				Map<String, Double> agg = new HashMap<>();
				agg.put("sum", sum);
				agg.put("bussiness", bussiness);
				agg.put("percent", percent);
				return agg;
			}
		};

	}
}

$ pip install PyGithub

from github import Github
import os
import codecs
def create(newfile):
    if not os.path.exists(newfile):
         f = open(newfile,'w')
         print newfile
         f.close()
    return
fileName='rct.txt'
respository='xaecbd/RCT'
g=Github("token",timeout=300)

repo=g.get_repo(respository)
stargazers=repo.get_stargazers_with_dates()
create(fileName)
with codecs.open(fileName,'a', 'utf-8') as f:
    for people in stargazers:
        print people.user.email
        if people.user.email is not None:
            if people.user.name is None:
                name = people.user.email
            else:
                name = people.user.name
            data=(name)+":"+(people.user.email)+"\r\n"
            f.write(data)

python crawer.py

/**
	 * 根据<master:slaves>获取执行分析任务ports规则
	 * 即获取其中一个slave,尽量保持均衡在不同机器上
	 * 
	 * @param clusterNodesMap
	 * @return <ip:ports>
	 */
	public static Map<String, Set<String>> generateAnalyzeRule(Map<String, List<String>> clusterNodesMap) {
		
		// 通过该map存储不同IP分配的数量，按照规则，优先分配数量最小的IP
		Map<String, Integer> staticsResult = new HashMap<>();
		Map<String, Set<String>> generateRule = new HashMap<>();

		// 此处排序是为了将slave数量最小的优先分配
		List<Map.Entry<String, List<String>>> sortList = new LinkedList<>(clusterNodesMap.entrySet());
		Collections.sort(sortList, new Comparator<Entry<String, List<String>>>() {
			@Override
			public int compare(Entry<String, List<String>> o1, Entry<String, List<String>> o2) {
				return o1.getValue().size() - o2.getValue().size();
			}
		});

		for (Entry<String, List<String>> entry : sortList) {
			List<String> slaves = entry.getValue();
			boolean isSelected = false;
			String tempPort = null;
			String tempIP = null;
			int num = 0;
			for (String slave : slaves) {
				String ip = slave.split(":")[0];
				String port = slave.split(":")[1];
				// 统计组里面不存在的IP优先分配
				if (!staticsResult.containsKey(ip)) {
					staticsResult.put(ip, 1);
					Set<String> generatePorts = generateRule.get(ip);
					if (generatePorts == null) {
						generatePorts = new HashSet<>();
					}
					generatePorts.add(port);
					generateRule.put(ip, generatePorts);
					isSelected = true;
					break;
				} else {
					// 此处是为了求出被使用最少的IP
					Integer staticsNum = staticsResult.get(ip);
					if (num == 0) {
						num = staticsNum;
						tempPort = port;
						tempIP = ip;
						continue;
					}
					if (staticsNum < num) {
						tempPort = port;
						tempIP = ip;
						num = staticsNum;
					}
				}

			}

			// 如果上面未分配,则选择staticsResult中数值最小的那个slave
			if (!isSelected) {
				if (slaves != null && slaves.size() > 0) {
					if (tempPort != null) {
						Set<String> generatePorts = generateRule.get(tempIP);
						if (generatePorts == null) {
							generatePorts = new HashSet<>();
						}
						generatePorts.add(tempPort);
						generateRule.put(tempIP, generatePorts);
						staticsResult.put(tempIP, staticsResult.get(tempIP) + 1);
					}
				}
			}
		}
		return generateRule;
	}

public class Jemalloc {

	private static long[] array16;

	private static long[] array192;

	private static long[] array768;

	private static long[] array4096;

	private static long[] array4194304;

	static {
		array16 = range(16, 128 + 1, 16);

		array192 = range(192, 512 + 1, 64);

		array768 = range(768, 4096 + 1, 256);

		array4096 = range(4096, 4194304 + 1, 4096);

		array4194304 = range(4194304, 536870912 + 1, 4194304);

	}

	/**
	 * 根据Jemalloc 估算分配内存大小 Small: All 2^n-aligned allocations of size 2^n will incur
	 * no additional overhead, due to how small allocations are aligned and packed.
	 * Small: [8], [16, 32, 48, ..., 128], [192, 256, 320, ..., 512], [768, 1024,
	 * 1280, ..., 3840] Large: The worst case size is half the chunk size, in which
	 * case only one allocation per chunk can be allocated. If the remaining
	 * (nearly) half of the chunk isn't otherwise useful for smaller allocations,
	 * the overhead will essentially be 50%. However, assuming you use a diverse
	 * mixture of size classes, the actual overhead shouldn't be a significant issue
	 * in practice. Large: [4 KiB, 8 KiB, 12 KiB, ..., 4072 KiB] Huge: Extra virtual
	 * memory is mapped, then the excess is trimmed and unmapped. This can leave
	 * virtual memory holes, but it incurs no physical memory overhead. Earlier
	 * versions of jemalloc heuristically attempted to optimistically map chunks
	 * without excess that would need to be trimmed, but it didn't save much system
	 * call overhead in practice. Huge: [4 MiB, 8 MiB, 12 MiB, ..., 512 MiB]
	 * 
	 * @param size
	 * @return
	 */
	public static long assign(long size) {
		if (size <= 4096) {
			// Small
			if (is_power2(size)) {
				return size;
			} else if (size < 128) {
				return min_ge(array16, size);
			} else if (size < 512) {
				return min_ge(array192, size);
			} else {
				return min_ge(array768, size);
			}
		} else if (size < 4194304) {
			// Large
			return min_ge(array4096, size);
		} else {
			// Huge
			return min_ge(array4194304, size);
		}
	}

	/**
	 * 创建一个long数组
	 * 
	 * @param start
	 * @param stop
	 * @param step
	 * @return
	 */
	public static long[] range(int start, int stop, int step) {
		int size = (stop - 1 - start) / step + 1;
		long[] array = new long[size];
		int index = 0;
		for (int i = start; i < stop; i = i + step) {
			array[index] = i;
			index++;
		}
		return array;
	}

	public static long min_ge(long[] srcArray, long key) {
		int index = binarySearch(srcArray, key);
		return srcArray[index];
	}

	// 二分查找最小值，即最接近要查找的值，但是要大于该值
	public static int binarySearch(long srcArray[], long key) {
		int mid = (0+srcArray.length-1) / 2;
		if (key == srcArray[mid]) {
			return mid;
		}

		if (key > srcArray[mid] && key <= srcArray[mid + 1]) {
			return mid + 1;
		}

		int start = 0;
		int end = srcArray.length - 1;
		while (start <= end) {
			mid = (end - start) / 2 + start;
			if (key == srcArray[mid]) {
				return mid;
			}
			if (key > srcArray[mid] && key <= srcArray[mid + 1]) {
				return mid + 1;
			}
			if (key < srcArray[mid]) {
				end = mid - 1;
			}
			if (key > srcArray[mid]) {
				start = mid + 1;
			}
		}
		return 0;
	}

	public static boolean is_power2(long size) {
		if (size == 0) {
			return false;
		}

		if ((size & (size - 1)) == 0) {
			return true;
		}

		return false;
	}
}

private static final long redisObject = (long)(8 + 8);
// 一个dictEntry，24字节，jemalloc会分配32字节的内存块
private static final long dicEntry = (long)(2 * 8 + 8 + 8);
private static final String patternString = "^[-\\\\+]?[\\\\d]*$";

private static long skiplistMaxLevel = 32;
private static long redisSharedInterges = 10000;
private static long longSize = 8;
private static long pointerSize = 8;

/**
 * 一个SDS结构占据的空间为：free所占长度+len所占长度+ buf数组的长度=4+4+len+1=len+9
 * 
 * @param length
 * @return
 */
private static long sds(long length) {
	long mem = 9 + length;
	return mem;
}

/**
 * 
 * 计算 string byte 大小
 * 
 * @param kv
 *            https://searchdatabase.techtarget.com.cn/wp-content/uploads/res/database/article/2011/2011-11-14-16-56-18.jpg
 * @return
 */
public static long CalculateString(KeyStringValueString kv) {
	long mem = KeyExpiryOverhead(kv);
	mem = dicEntry + SizeofString(kv.getRawKey());
	mem = mem + redisObject + SizeofString(kv.getValueAsString());
	return mem;
}

public static long CalculateLinkedList(KeyStringValueList kv) {
	long mem = KeyExpiryOverhead(kv);
	mem = mem + SizeofString(kv.getRawKey());
	mem = mem + redisObject;
	mem = mem + dicEntry;
	long length = kv.getValueAsStringList().size();
	mem = mem + LinkedListEntryOverhead() * length;
	mem = mem + LinkedlistOverhead();
	mem = mem + redisObject * length;
	for (String value : kv.getValueAsStringList()) {
		mem = mem + SizeofString(value);
	}

	return mem;
}

public static long CalculateZipList(KeyStringValueList kv) {
	long mem = KeyExpiryOverhead(kv);
	mem = mem + dicEntry;
	mem = mem + SizeofString(kv.getRawKey());
	mem = mem + redisObject;
	long length = kv.getValueAsStringList().size();
	mem = mem + ZiplistOverhead(length);
       
	for (String value : kv.getValueAsStringList()) {
		mem = mem + ZiplistAlignedStringOverhead(value);
	}

	return mem;
}

public static long CalculateHash(KeyStringValueHash kv) {
	long mem = KeyExpiryOverhead(kv);
	mem = mem + SizeofString(kv.getRawKey());
	mem = mem + redisObject;
	mem = mem + dicEntry;
	long length = kv.getValueAsHash().size();
	mem = mem + HashtableOverhead(length);

	for (String key : kv.getValueAsHash().keySet()) {
		String value = kv.getValueAsHash().get(key);
		mem = mem + SizeofString(key);
		mem = mem + SizeofString(value);
		mem = mem + 2 * redisObject;
		mem = mem + HashtableEntryOverhead();
	}

	return mem;
}

public static long CalculateSet(KeyStringValueSet kv) {
	long mem = KeyExpiryOverhead(kv);

	mem = mem + SizeofString(kv.getRawKey());
	mem = mem + redisObject;
	mem = mem + dicEntry;
	long length = kv.getValueAsSet().size();
	mem = mem + HashtableOverhead(length);
	mem = mem + redisObject * length;

	for (String value : kv.getValueAsSet()) {
		mem = mem + SizeofString(value);
		mem = mem + 2 * redisObject;
		mem = mem + HashtableEntryOverhead();
	}

	return mem;
}

public static long CalculateIntSet(KeyStringValueSet kv) {
	long mem = KeyExpiryOverhead(kv);
	mem = mem + dicEntry;
	mem = mem + SizeofString(kv.getRawKey());
	mem = mem + redisObject;
	
	long length = kv.getValueAsSet().size();
	mem = mem + IntsetOverhead(length);

	for (String value : kv.getValueAsSet()) {
		mem = mem + ZiplistAlignedStringOverhead(value);
	}

	return mem;
}

public static long CalculateZSet(KeyStringValueZSet kv) {
	long mem = KeyExpiryOverhead(kv);

	mem = mem + SizeofString(kv.getRawKey());
	mem = mem + redisObject;
	mem = mem + dicEntry;
	long length = kv.getValueAsSet().size();
	mem = mem + SkiplistOverhead(length);
	mem = mem + redisObject * length;

	for (ZSetEntry value : kv.getValueAsZSet()) {
		mem = mem + 8;
		mem = mem + SizeofString(value.getElement());
		// TODO 还有个 score
		mem = mem + 2 * redisObject;
		mem = mem + SkiplistEntryOverhead();
	}

	return mem;
}

// TopLevelObjOverhead get memory use of a top level object
// Each top level object is an entry in a dictionary, and so we have to include
// the overhead of a dictionary entry
public static long TopLevelObjOverhead() {
	return HashtableEntryOverhead();
}

/**
 * SizeofString get memory use of a string
 * https://github.com/antirez/redis/blob/unstable/src/sds.h
 * 
 * @param bytes
 * @return
 */
public static long SizeofString(byte[] bytes) {
	String value = new String(bytes);

	if (isInteger(value)) {
		try {
			Long num = Long.parseLong(value);
			if (num < redisSharedInterges && num > 0) {
				return 0;
			}
			return 8;
		} catch (NumberFormatException e) {
		}
	}

	return Jemalloc.assign(sds(bytes.length));
}


public static long SizeofString(String value) {
	if (isInteger(value)) {
		try {
			Long num = Long.parseLong(value);
			if (num < redisSharedInterges && num > 0) {
				return 0;
			}
			return 8;
		} catch (NumberFormatException e) {
		}
	}

	return Jemalloc.assign(sds(value.length()));
}

public static long DictOverhead(long size) {
	 return Jemalloc.assign(56 + 2*pointerSize + nextPower(size) * 3*8);
}

public static boolean isInteger(String str) {
	Pattern pattern = Pattern.compile(patternString);
	return pattern.matcher(str).matches();
}

/**
 * 过期时间也是存储为一个 dictEntry，时间戳为 int64；
 * 
 * @param kv
 * @return
 */
// KeyExpiryOverhead get memory useage of a key expiry
// Key expiry is stored in a hashtable, so we have to pay for the cost of a
// hashtable entry
// The timestamp itself is stored as an int64, which is a 8 bytes
@SuppressWarnings("rawtypes")
public static long KeyExpiryOverhead(KeyValuePair kv) {
	// If there is no expiry, there isn't any overhead
	if (kv.getExpiredType() == ExpiredType.NONE) {
		return 0;
	}
	return HashtableEntryOverhead() + 8;
}

public static long HashtableOverhead(long size) {
	return 4 + 7 * longSize + 4 * pointerSize + nextPower(size) * pointerSize * 3 / 2;
}

// HashtableEntryOverhead get memory use of hashtable entry
// See https://github.com/antirez/redis/blob/unstable/src/dict.h
// Each dictEntry has 2 pointers + int64
public static long HashtableEntryOverhead() {
	return 2 * pointerSize + 8;
}

public static long ZiplistOverhead(long size) {
	return Jemalloc.assign(12 + 21 * size);
}

public static long ZiplistAlignedStringOverhead(String value) {
	try {
		Long.parseLong(value);
		return 8;
	} catch (NumberFormatException e) {
	}
	return Jemalloc.assign(value.length());
}

// LinkedlistOverhead get memory use of a linked list
// See https://github.com/antirez/redis/blob/unstable/src/adlist.h
// A list has 5 pointers + an unsigned long
public static long LinkedlistOverhead() {
	return longSize + 5 * pointerSize;
}

// LinkedListEntryOverhead get memory use of a linked list entry
// See https://github.com/antirez/redis/blob/unstable/src/adlist.h
// A node has 3 pointers
public static long LinkedListEntryOverhead() {
	return 3 * pointerSize;
}

// SkiplistOverhead get memory use of a skiplist
public static long SkiplistOverhead(long size) {
	return 2 * pointerSize + HashtableOverhead(size) + (2 * pointerSize + 16);
}

// SkiplistEntryOverhead get memory use of a skiplist entry
public static long SkiplistEntryOverhead() {
	return HashtableEntryOverhead() + 2 * pointerSize + 8 + (pointerSize + 8) * zsetRandLevel();
}

public static long nextPower(long size) {
	long power = 1;
	while (power <= size) {
		power = power << 1;
	}
	return power;
}

public static long zsetRandLevel() {
	long level = 1;
	int rint = new Random().nextInt(65536);
	int flag = 65535 / 4;
	while (rint < flag) {// skiplistP
		level++;
		rint = new Random().nextInt(65536);
	}
	if (level < skiplistMaxLevel) {
		return level;
	}
	return skiplistMaxLevel;
}

public static long  IntsetOverhead(long size) {
    //     typedef struct intset {
    //     uint32_t encoding;
    //     uint32_t length;
    //     int8_t contents[];
    //      } intset;
    return (4 + 4) * size;
}