kafka之CopyOnWriteMap

- 日理万妓 2023-01-21 14:21 18阅读 0赞

# 一：kafka之CopyOnWriteMap #

kafka中每个分区的数据都是读多写少的操作，通过内存可见性提高每个分区读写性能。

CopyOnWriteMap实体类：

/*
     * Licensed to the Apache Software Foundation (ASF) under one or more
     * contributor license agreements. See the NOTICE file distributed with
     * this work for additional information regarding copyright ownership.
     * The ASF licenses this file to You under the Apache License, Version 2.0
     * (the "License"); you may not use this file except in compliance with
     * the License. You may obtain a copy of the License at
     *
     *    http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package org.apache.kafka.common.utils;
    
    import java.util.Collection;
    import java.util.Collections;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Set;
    import java.util.concurrent.ConcurrentMap;
    
    /**
     * A simple read-optimized map implementation that synchronizes only writes and does a full copy on each modification
     */
    public class CopyOnWriteMap<K, V> implements ConcurrentMap<K, V> {
    
        /**
         * todo 核心变量就是一个map
         * 用volatile 保证内存可见性
         */
        private volatile Map<K, V> map;
    
        public CopyOnWriteMap() {
            this.map = Collections.emptyMap();
        }
    
        public CopyOnWriteMap(Map<K, V> map) {
            this.map = Collections.unmodifiableMap(map);
        }
    
        @Override
        public boolean containsKey(Object k) {
            return map.containsKey(k);
        }
    
        @Override
        public boolean containsValue(Object v) {
            return map.containsValue(v);
        }
    
        @Override
        public Set<java.util.Map.Entry<K, V>> entrySet() {
            return map.entrySet();
        }
    
        @Override
        public V get(Object k) {
            return map.get(k);
        }
    
        @Override
        public boolean isEmpty() {
            return map.isEmpty();
        }
    
        @Override
        public Set<K> keySet() {
            return map.keySet();
        }
    
        @Override
        public int size() {
            return map.size();
        }
    
        @Override
        public Collection<V> values() {
            return map.values();
        }
    
        @Override
        public synchronized void clear() {
            this.map = Collections.emptyMap();
        }
    
        @Override
        public synchronized V put(K k, V v) {
            Map<K, V> copy = new HashMap<K, V>(this.map);
            V prev = copy.put(k, v);
            this.map = Collections.unmodifiableMap(copy);
            return prev;
        }
    
        @Override
        public synchronized void putAll(Map<? extends K, ? extends V> entries) {
            Map<K, V> copy = new HashMap<K, V>(this.map);
            copy.putAll(entries);
            this.map = Collections.unmodifiableMap(copy);
        }
    
        @Override
        public synchronized V remove(Object key) {
            Map<K, V> copy = new HashMap<K, V>(this.map);
            V prev = copy.remove(key);
            this.map = Collections.unmodifiableMap(copy);
            return prev;
        }
    
        @Override
        public synchronized V putIfAbsent(K k, V v) {
            if (!containsKey(k))
                return put(k, v);
            else
                return get(k);
        }
    
        @Override
        public synchronized boolean remove(Object k, Object v) {
            if (containsKey(k) && get(k).equals(v)) {
                remove(k);
                return true;
            } else {
                return false;
            }
        }
    
        @Override
        public synchronized boolean replace(K k, V original, V replacement) {
            if (containsKey(k) && get(k).equals(original)) {
                put(k, replacement);
                return true;
            } else {
                return false;
            }
        }
    
        @Override
        public synchronized V replace(K k, V v) {
            if (containsKey(k)) {
                return put(k, v);
            } else {
                return null;
            }
        }
    
    }

这个类其实最关键的就是volatile关键字

RecordAccumulator实体类中有批次属性：

private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;
     //初始化的时候 CopyOnWriteMap 
    this.batches = new CopyOnWriteMap<>();