sync.Map使用和介绍

1、首先看下该sync.Map的使用：



package main
import (
    "sync"
    "fmt"
)

func main() {
    //开箱即用
    var sm sync.Map
    //store 方法,添加元素
    sm.Store(1,"a")
    //Load 方法，获得value
    if v,ok:=sm.Load(1);ok{
        fmt.Println(v)
    }
    //LoadOrStore方法，获取或者保存
    //参数是一对key：value，如果该key存在且没有被标记删除则返回原先的value（不更新）和true；不存在则store，返回该value 和false
    if vv,ok:=sm.LoadOrStore(1,"c");ok{
        fmt.Println(vv)
    }
    if vv,ok:=sm.LoadOrStore(2,"c");!ok{
        fmt.Println(vv)
    }
    //遍历该map，参数是个函数，该函数参的两个参数是遍历获得的key和value，返回一个bool值，当返回false时，遍历立刻结束。
    sm.Range(func(k,v interface{})bool{
        fmt.Print(k)
        fmt.Print(":")
        fmt.Print(v)
        fmt.Println()
        return true
    })
}

运行结果：



a
a
c
1:a
2:c

2、利用传统的sync.RWMutex+Map 实现并发安全的map：



var rwmap = struct{
    sync.RWMutex
    m map[string]string
}{m: make(map[string]sring)}

读数据时候，读锁锁定：



rwmap.RLock()
value:= rwmap.m["key"]
rwmap.RUnlock()
fmt.Println("key:", value)

写数据的时候，写锁锁定:



rwmap.Lock()
rwmap.m["key"]="value"
rwmap.Unlock()

3、两种map的性能对比： 

下面是有人做的两种map性能对比图： 

 

        可见随着cpu核心数的增加、并发加剧，这种读写锁+map的方式性能在不停的衰减，并且在核数为4的时候出现了性能的拐点；而sync.Map虽然性能不是特别好，但是相对比较平稳。 

  4、sync.Map 源码解析 

  源码位于：src\sync\map.go 

  首先查看一下sync.Map的数据结构：



  type Map struct {
    // 该锁用来保护dirty
    mu Mutex
    // 存读的数据，因为是atomic.value类型，只读类型，所以它的读是并发安全的
    read atomic.Value // readOnly
    //包含最新的写入的数据，并且在写的时候，会把read 中未被删除的数据拷贝到该dirty中，因为是普通的map存在并发安全问题，需要用到上面的mu字段。
    dirty map[interface{}]*entry
    // 从read读数据的时候，会将该字段+1，当等于len（dirty）的时候，会将dirty拷贝到read中（从而提升读的性能）。
    misses int
}


read的数据结构是：



type readOnly struct {
    m  map[interface{}]*entry
    // 如果Map.dirty的数据和m 中的数据不一样是为true
    amended bool 
}

entry的数据结构：



type entry struct {
    //可见value是个指针类型，虽然read和dirty存在冗余情况（amended=false），但是由于是指针类型，存储的空间应该不是问题
    p unsafe.Pointer // *interface{}
}

Delete 

首先来看delete方法



func (m *Map) Delete(key interface{}) {
    read, _ := m.read.Load().(readOnly)
    e, ok := read.m[key]
    //如果read中没有，并且dirty中有新元素，那么就去dirty中去找
    if !ok && read.amended {
        m.mu.Lock()
        //这是双检查（上面的if判断和锁不是一个原子性操作）
        read, _ = m.read.Load().(readOnly)
        e, ok = read.m[key]
        if !ok && read.amended {
            //直接删除
            delete(m.dirty, key)
        }
        m.mu.Unlock()
    }
    if ok {
    //如果read中存在该key，则将该value 赋值nil（采用标记的方式删除！）
        e.delete()
    }
}
func (e *entry) delete() (hadValue bool) {
    for {
        p := atomic.LoadPointer(&e.p)
        if p == nil || p == expunged {
            return false
        }
        if atomic.CompareAndSwapPointer(&e.p, p, nil) {
            return true
        }
    }
}

Store 

新加元素



func (m *Map) Store(key, value interface{}) {
    // 如果m.read存在这个key，并且没有被标记删除，则尝试更新。
    read, _ := m.read.Load().(readOnly)
    if e, ok := read.m[key]; ok && e.tryStore(&value) {
        return
    }
    // 如果read不存在或者已经被标记删除
    m.mu.Lock()
    read, _ = m.read.Load().(readOnly)
    if e, ok := read.m[key]; ok {
    //如果entry被标记expunge，则表明dirty没有key，可添加入dirty，并更新entry
        if e.unexpungeLocked() { 
            //加入dirty中
            m.dirty[key] = e
        }
        //更新value值
        e.storeLocked(&value) 
        //dirty 存在该key，更新
    } else if e, ok := m.dirty[key]; ok { 
        e.storeLocked(&value)
        //read 和dirty都没有，新添加一条
    } else {
     //dirty中没有新的数据，往dirty中增加第一个新键
        if !read.amended { 
            //将read中未删除的数据加入到dirty中
            m.dirtyLocked() 
            m.read.Store(readOnly{m: read.m, amended: true})
        }
        m.dirty[key] = newEntry(value) 
    }
    m.mu.Unlock()
}

//将read中未删除的数据加入到dirty中
func (m *Map) dirtyLocked() {
    if m.dirty != nil {
        return
    }
    read, _ := m.read.Load().(readOnly)
    m.dirty = make(map[interface{}]*entry, len(read.m))
    //read如果较大的话，可能影响性能
    for k, e := range read.m {
    //通过此次操作，dirty中的元素都是未被删除的，可见expunge的元素不在dirty中
        if !e.tryExpungeLocked() {
            m.dirty[k] = e
        }
    }
}
//判断entry是否被标记删除，并且将标记为nil的entry更新标记为expunge
func (e *entry) tryExpungeLocked() (isExpunged bool) {
    p := atomic.LoadPointer(&e.p)
    for p == nil {
        // 将已经删除标记为nil的数据标记为expunged
        if atomic.CompareAndSwapPointer(&e.p, nil, expunged) {
            return true
        }
        p = atomic.LoadPointer(&e.p)
    }
    return p == expunged
}
//对entry 尝试更新
func (e *entry) tryStore(i *interface{}) bool {
    p := atomic.LoadPointer(&e.p)
    if p == expunged {
        return false
    }
    for {
        if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
            return true
        }
        p = atomic.LoadPointer(&e.p)
        if p == expunged {
            return false
        }
    }
}
//read里 将标记为expunge的更新为nil
func (e *entry) unexpungeLocked() (wasExpunged bool) {
    return atomic.CompareAndSwapPointer(&e.p, expunged, nil)
}
//更新entry
func (e *entry) storeLocked(i *interface{}) {
    atomic.StorePointer(&e.p, unsafe.Pointer(i))
}

可见，每次操作先检查read，因为read 并发安全，性能好些；read不满足，则加锁检查dirty，一旦是新的键值，dirty会被read更新。 

Load 

加载方法，查找key。



func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
    //因read只读，线程安全，先查看是否满足条件
    read, _ := m.read.Load().(readOnly)
    e, ok := read.m[key]
    //如果read没有，并且dirty有新数据，那从dirty中查找，由于dirty是普通map，线程不安全，这个时候用到互斥锁了
    if !ok && read.amended {
        m.mu.Lock()
        // 双重检查
        read, _ = m.read.Load().(readOnly)
        e, ok = read.m[key]
        // 如果read中还是不存在，并且dirty中有新数据
        if !ok && read.amended {
            e, ok = m.dirty[key]
            // mssLocked（）函数是性能是sync.Map 性能得以保证的重要函数，目的讲有锁的dirty数据，替换到只读线程安全的read里
            m.missLocked()
        }
        m.mu.Unlock()
    }
    if !ok {
        return nil, false
    }
    return e.load()
}
//dirty 提升至read 关键函数，当misses 经过多次因为load之后，大小等于len（dirty）时候，讲dirty替换到read里，以此达到性能提升。
func (m *Map) missLocked() {
    m.misses++
    if m.misses < len(m.dirty) {
        return
    }
    //原子操作，耗时很小
    m.read.Store(readOnly{m: m.dirty})
    m.dirty = nil
    m.misses = 0
}

源码用的是1.9版本，通过阅读源码我们发现sync.Map是通过冗余的两个数据结构(read、dirty),实现性能的提升。为了提升性能，load、delete、store等操作尽量使用只读的read；为了提高read的key击中概率，采用动态调整，将dirty数据提升为read；对于数据的删除，采用延迟标记删除法，只有在提升dirty的时候才删除。

java中的map有好几种输出方法，本篇博客只讲其中的迭代器输出
代码如下
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class MapOutput {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Map<Integer,String> map = new HashMap<Integer,String>();
		map.put(1, "aaa");
		map.put(2, "bbb");
		map.put(3, "ccc");
		map.put(4, "ddd");
		
		Iterator<Map.Entry<Integer,String>> it=map.entrySet().iterator();
		while(it.hasNext()){
			Map.Entry<Integer,String> entry=it.next();
			System.out.println("key="+entry.getKey()+","+"value="+entry.getValue());
		}
		

	}

}


输出结果如下
key=1,value=aaa

key=2,value=bbb

key=3,value=ccc

key=4,value=ddd

map使用return false无法退出循环

解决方法：

使用for替换map，for可以使用return false退出循环

Map结构是一种非常常见的结构，在各种程序语言都有对应的api，由于Spark的底层语言是Scala，所以有必要来了解下Scala中的Map使用方法。

（1）不可变Map

特点：

api不太丰富

如果是var修饰，引用可变，支持读写

如果是val修饰，引用不可变，只能写入一次值，其后只读
var a:Map[String,Int]=Map("k1"->1,"k2"->2)//初始化构造函数
    a += ("k3"->3)//添加元素
    a += ("k4"->4)//添加元素
    a += ("k1"->100)//已经存在添加元素会覆盖
    a -= ("k2","k1")//删除元素
//    a("k1") = "foo"//不支持
    println(a.contains("k6"))//是否包含某元素
    println(a.size)//打印大小
    println(a.get("k1").getOrElse("default")) //根据key读取元素，不存在就替换成默认值
    a.foreach{case (e,i) => println(e,i)} //遍历打印1
    for( (k,v)<-a ) println(k,v) //遍历打印2
    println(a.isEmpty)//判断是否为空
    a.keys.foreach(println)//只打印key
    a.values.foreach(println)//只打印value
    a=Map()//数据清空使用再次new
    println(a.size)
    a.toSeq.sortBy(_._1)//升序排序 key
    a.toSeq.sortBy(_._2)//升序排序 value
    a.toSeq.sortWith(_._1>_._1) //降序排序 key
    a.toSeq.sortWith(_._2>_._2) //降序排序 value
    
    //下面自定义按英文字母或数字排序
    implicit  val KeyOrdering=new Ordering[String] {
      override def compare(x: String, y: String): Int = {
        x.compareTo(y)
      }
    }
    println(a.toSeq.sorted)

（2）可变Map例子

特点：

api丰富与Java中Map基本类似

如果是var修饰，引用可变，支持读写

如果是val修饰，引用不可变，支持读写
def map3(): Unit ={
  //不可变Map+var关键词修饰例子
  var a:scala.collection.mutable.Map[String,Int]=scala.collection.mutable.Map("k1"->1,"k2"->2)//初始化构造函数
  a += ("k3"->3)//添加元素
  a += ("k4"->4)//添加元素
  a += ("k1"->100)//已经存在添加元素会覆盖
  a += ("k1"->100,"k9"->9)//添加多个元素
  a -= ("k2","k1")//删除元素
  a ++= List("CA" -> 23, "CO" -> 25)//追加集合
  a --= List("AL", "AZ")//删除集合

  a.retain((k,v)=> k=="k1")//只保留等于k1元素，其他的删除
  a.put("put1",200)//put
  a.remove("k2")//remove
  a.clear()//清空
  a("k3")=100//支持

  println(a.contains("k6"))//是否包含某元素
  println(a.size)//打印大小
  println(a.get("k1").getOrElse("default")) //根据key读取元素，不存在就替换成默认值
  a.foreach{case (e,i) => println(e,i)} //遍历打印1
  for( (k,v)<-a ) println(k,v) //遍历打印2
  println(a.isEmpty)//判断是否为空
  a.keys.foreach(println)//只打印key
  a.values.foreach(println)//只打印value
  a=scala.collection.mutable.Map()//引用能变
  println(a.size)
  a.toSeq.sortBy(_._1)//排序 key
  a.toSeq.sortBy(_._2)//排序 value
  a.toSeq.sortWith(_._1>_._1) //降序排序 key
  a.toSeq.sortWith(_._2>_._2) //降序排序 value
  
      //下面自定义按英文字母或数字排序
  implicit  val KeyOrdering=new Ordering[String] {
    override def compare(x: String, y: String): Int = {
      x.compareTo(y)
    }
  }
  println(a.toSeq.sorted)
}



















有什么问题可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。 技术债不能欠，健康债更不能欠， 求道之路，与君同行。