AdminClient adminClient = KafkaAdminClient.create(props);

ListTopicsResult listTopicsResult = adminClient.listTopics();
Set<String> topics = listTopicsResult.names().get();
System.out.println(topics);

DescribeTopicsResult describeTopicsResult =
adminClient.describeTopics(Arrays.asList("tpc_4", "tpc_3"));
Map<String, TopicDescription> res = describeTopicsResult.all().get();
Set<String> ksets = res.keySet();
for (String k : ksets) {
  System.out.println(res.get(k));
}

// 参数配置
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"doit01:9092,doit02:9092,doit03:
9092");
props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);
// 创建 admin client 对象
AdminClient adminClient = KafkaAdminClient.create(props);
// 由服务端 controller 自行分配分区及副本所在 broker
NewTopic tpc_3 = new NewTopic("tpc_3", 2, (short) 1);
// 手动指定分区及副本的 broker 分配
HashMap<Integer, List<Integer>> replicaAssignments = new HashMap<>();
// 分区 0，分配到 broker0，broker1
replicaAssignments.put(0,Arrays.asList(0,1));
// 分区 1，分配到 broker0,broker2
replicaAssignments.put(1,Arrays.asList(0,1));
NewTopic tpc_4 = new NewTopic("tpc_4", replicaAssignments);
CreateTopicsResult result = adminClient.createTopics(Arrays.asList(tpc_3,tpc_4));
// 从 future 中等待服务端返回
try {
  result.all().get();
} catch (Exception e) {
  e.printStackTrace();
}
adminClient.close();

DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList("tpc_1",
"tpc_1"));
Map<String, KafkaFuture<Void>> values = deleteTopicsResult.values();
System.out.println(values);

public class consumer{
    public static void main(String[] args){
        ConcurrentHashMap<Long, String> guidMap = new ConcurrentHashMap<>();
        // 启动数据消费线程
        new Thread(new ConsumeRunnable(guidMap)).start();
        // 启动统计及消费输出结果的线程（每5s输出一次）
        // 优雅一点实现定时调度，可以用各种定时调度器（有第三方的，也可以用JDK自己的，Timer）
        Timer timer = new Timer();
        timer.scheduleAtFixedRate(new StatisticTask(guidMap),0,1000);
    }
}
class ConsumeRunnable implements Runnable{
    ConcurrentHashMap<Long, String> guidMap;
    public ConsumeRunnable(ConcurrentHashMap<Long, String> guidMap){
        this.guidMap = guidMap;
    }
    public void run(){
        Properties props = new Properties();
        // 定义 kakfa 服务的地址，不需要将所有 broker 指定上
    props.put("bootstrap.servers", "doitedu01);
        // key 的反序列化类
    props.put("key.deserializer",
                  "org.apache.kafka.common.serialization.StringDeserializer");
    // value 的反序列化类
    props.put("value.deserializer",
          "org.apache.kafka.common.serialization.StringDeserializer");
        // 如果没有消费偏移量记录，则自动重设为起始 offset：latest, earliest, none
    props.put("auto.offset.reset","latest");
        // 制定 consumer group
    props.put("group.id", "g1");
        while (true) {
      // 读取数据，读取超时时间为 5000ms
      ConsumerRecords<String, String> records = consumer.poll(5000);
      for (ConsumerRecord<String, String> record : records)
        String eventJson = record.value();
              UserEvent userEvent = JSON.parseObject(eventJson,UserEvent.class);
              guidMap.put(userEvent.getGuid(), "");
      }
      }
}
class StatisticTask extends TimerTask{
    ConcurrentHashMap<Long, String> guidMap;
    public StatisticTask(ConcurrentHashMap<Long, String> guidMap){
        this.guidMap = guidMap;
    }
    public void run(){
        System.out.println("截止到当前的用户总数为："+guidMap.size());
    }
}

  public V put(K key, V value) {
        return putVal(key, value, false);
        // onlyIfAbsent如果是false，那么每次都会用新值替换掉旧值
    }
    final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        // 其中 spread 方法会综合高位低位, 具有更好的 hash 性
        int hash = spread(key.hashCode());
        int binCount = 0;
        // 死循环
        for (Node<K,V>[] tab = table;;) {
            // f 是链表头节点
            // fh 是链表头结点的 hash
            // i 是链表在 table 中的下标
            Node<K,V> f; int n, i, fh;
            // 要创建 table
            if (tab == null || (n = tab.length) == 0)
                // 初始化 table 使用了 cas, 无需 synchronized 创建成功, 进入下一轮循环
                // 因为是懒惰初始化的，所以直到现在才开始创建 初始化使用cas 创建，其它失败得再次进入循环，没有用syn 我们得线程并没有被阻塞住
                tab = initTable();
                // 要创建链表头节点
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                // 添加链表头使用了 cas, 无需 synchronized
                // 用cas将头节点加进去，如果加入失败了，继续循环
                if (casTabAt(tab, i, null,
                        new Node<K,V>(hash, key, value, null)))
                    break;
            }
            // 帮忙扩容
            // 其实就是看你的头结点是不是 ForwardingNode，其对应得MOVED是一个负数
            else if ((fh = f.hash) == MOVED)
                // 帮忙之后, 进入下一轮循环
                // 锁住当前的链表，帮助去扩容
                tab = helpTransfer(tab, f);
            // 能进入这个else，说明 table既不处于扩容中，也不是处于table的初始化过程中，而且这时肯定发生了锁下标的冲突
            else {
                V oldVal = null;
                // 锁住链表头节点
                // 并没有锁住整个tab，而是锁住这个桶链表的头节点
                synchronized (f) {
                    // 再次确认链表头节点没有被移动
                    if (tabAt(tab, i) == f) {
                        // 链表
                        // 链表的头节点hash码大于等于 0 
                        if (fh >= 0) {
                            binCount = 1;
                            // 遍历链表
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
// 找到相同的 key
                                if (e.hash == hash &&
                                        ((ek = e.key) == key ||
                                                (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    // 更新
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                // 已经是最后的节点了, 新增 Node, 追加至链表尾
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                            value, null);
                                    break;
                                }
                            }
                        }
                        // 红黑树
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            // putTreeVal 会看 key 是否已经在树中, 是, 则返回对应的 TreeNode
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                    value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                    // 释放链表头节点的锁
                }
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        // 如果链表长度 >= 树化阈值(8), 进行链表转为红黑树
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 增加 size 计数
        addCount(1L, binCount);
        return null;
    }
    private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        // 这个hash有没有被创建
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)
                // 让出cpu的使用权，如果cpu的时间片没有其它线程了，那么还是会分给这个线程，只是让他不至于充分利用cpu，少占用一点cpu的时间。
                Thread.yield();
                // 尝试将 sizeCtl 设置为 -1（表示初始化 table）
            // 而其它的线程，再次进入循环，首先 不小于0了，其次，之前的 sc也已经变了，cas失败，再次循环的时候，发现 tab已经不为空了，结束循环
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                // 获得锁, 创建 table, 这时其它线程会在 while() 循环中 yield 直至 table 创建
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        // 计算出下一次要扩容的阈值
                        sc = n - (n >>> 2);
                    }
                } finally {
                    // 计算出下一次要扩容的阈值
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }
    // check 是之前 binCount 的个数
  // 运用了 longadder 的思想
    private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        if (
            // 已经有了 counterCells, 向 cell 累加
            // 累加单元数组不为空
                (as = counterCells) != null ||
                        // 还没有, 向 baseCount 累加
            // 一个基础数值累加
                        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)
        ) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (
                // 还没有 counterCells
                    as == null || (m = as.length - 1) < 0 ||
                            // 还没有 cell
                            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                            // cell cas 增加计数失败
                            !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
            ) {
                // 创建累加单元数组和cell, 累加重试
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            // 获取元素个数
            s = sumCount();
        }
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            // 看看元素的个数是否大于扩容的阈值
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                    (n = tab.length) < MAXIMUM_CAPACITY) {
                int rs = resizeStamp(n);
                if (sc < 0) {
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                            sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                            transferIndex <= 0)
                        break;
                    // newtable 已经创建了，帮忙扩容
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        // 首次调用，因为是懒惰初始化的，所以还没有创建
                        transfer(tab, nt);
                }
                // 需要扩容，这时 newtable 未创建
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                        (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);
                s = sumCount();
            }
        }
    }