什么是幂等性？

对于同一笔业务操作，不管调用多少次，得到的结果都是一样的。普通方式 只适合单机

jvm加锁方式Lock只能在一个jvm中起效，如果多个请求都被同一套系统处理，上面这种使用Lock的方式是没有问题的，不过互联网系统中，多数是采用集群方式部署系统，同一套代码后面会部署多套，如果支付宝同时发来多个通知经过负载均衡转发到不同的机器，上面的锁就不起效了。此时对于多个请求相当于无锁处理了

3. 悲观锁方式，使用数据库中悲观锁实现。悲观锁类似于方式二中的Lock, 只不过是依靠数据库来实现的。数据库中悲观锁使用for update来实现的方式3可以保证接口的幂等性，不过存在缺点，如果业务耗时，并发情况下，后面线程会长期处于等待状态，占用很多线程，让这些线程处于无效等待状态。

4. 乐观锁方式，使用数据库中乐观锁实现。主要还是通过版本号，类似于java的cas操作，那么问题来了，在多线程的情况下，一个线程取得数据了，其它线程会怎么样呢？会不断重试，不会阻塞，可以理解为自旋，阻塞或唤醒一个Java线程需要操作系统切换CPU状态来完成，这种状态转换需要耗费处理器时间。如果同步代码块中的内容过于简单，状态转换消耗的时间有可能比用户代码执行的时间还要长。

在许多场景中，同步资源的锁定时间很短，为了这一小段时间去切换线程，线程挂起和恢复现场的花费可能会让系统得不偿失。如果物理机器有多个处理器，能够让两个或以上的线程同时并行执行，我们就可以让后面那个请求锁的线程不放弃CPU的执行时间，看看持有锁的线程是否很快就会释放锁。

而为了让当前线程“稍等一下”，我们需让当前线程进行自旋，如果在自旋完成后前面锁定同步资源的线程已经释放了锁，那么当前线程就可以不必阻塞而是直接获取同步资源，从而避免切换线程的开销。这就是自旋锁。

5. 唯一约束

常见方式：悲观锁，乐观锁，唯一约束

最优排序：乐观锁 > 唯一约束 > 悲观锁

乐观锁CAS的用法，比方说在couchbase中，key是自带版本号的

object OriginCasStreamingTransfer {

implicit val conf = ConfigFactory.load

def main(args: Array[String]): Unit = {

val cbSingleUerThreshold = conf.getInt("cb_single_user_threshold.size")

val batchSize = conf.getInt("batch.size") // 消息批次处理

val zkQuorum = conf.getString("test_kafka.zkQuorum")

val brokers = conf.getString("test_kafka.brokers")

val topics = conf.getString("test_kafka.topics")

val group = conf.getString("test_kafka.group.id")

val seconds: Long = 120 // spark streaming mini batch time interval 防止pending batch过多造成流任务失败

val numThreads = 16

val topicMap = topics.split(",").map((_, numThreads)).toMap

val kafkaParams = Map(

("metadata.broker.list", brokers),

("zookeeper.connect", zkQuorum),

("auto.offset.reset", "smallest"),

("zookeeper.connection.timeout.ms", "30000"),

("group.id", group)

)

val sparkConf = new SparkConf()

.setIfMissing("spark.master", "local[*]")

.setAppName("CasStreamingTransfer")

.set("spark.default.parallelism", "1000")

.set("spark.executor.memory", "8g")

val spark = SparkSession

.builder()

.appName("CasStreamingTransfer")

.config(sparkConf)

.getOrCreate()

val ssc = new StreamingContext(spark.sparkContext, Seconds(seconds))

val stream = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](

ssc, kafkaParams, topicMap,

StorageLevel.MEMORY_AND_DISK_SER) // key value key解码格式 value解码格式

stream.foreachRDD{each_rdd =>

each_rdd.repartition(1000).foreachPartition{ each_partition =>

while (each_partition.hasNext){

var batchCount = 0

val batchRecord = ListBuffer[String]()

while (batchCount < batchSize && each_partition.hasNext){

val item = each_partition.next()

batchRecord += item._1

batchCount += 1

}

val errorReadResult = collection.mutable.Set[String]()

val readResult = CbManager.getInstance().readCasAsync("test_playback_record_couchbase", batchRecord.asJava, errorReadResult.asJava)

// println("readResult: " + readResult)

val writeResult = collection.mutable.Map[CbCasEntity, Array[Byte]]()

for ((key, value)

val uid = key.getId // 取出的用户uid

// println("uid: " + uid)

val cas = key.getCas // 取出的用户的couchbase当中的cas值

val parseResult = PlayRecord.UserPlayRecords.parseFrom(value) // 解析couchbase当中的value字段

// 解析byte[], 拆分出旧，新的播放记录

val transferCount = calTransferCount(parseResult, cbSingleUerThreshold) // 计算需要从cb转移到hikv旧的播放记录量

val oldCbCutEntity = cutOldPlayRecordsFromCb(transferCount, parseResult) // 截取couchbase中比较旧的播放记录

val newCbUserPlayRecords = cutNewPlayRecordsFromCb(oldCbCutEntity.getTimeRecordBreakCount, oldCbCutEntity.getRecordEntryCount, parseResult) // 截取couchbase中比较新的播放记录

// println("transferCount: " + transferCount)

try{

val playRecordData = PlayRecordDataDao.getInstance().get(uid).toCompletableFuture.get // 阻塞读取

// println(playRecordData.getPlayRecord)

if (playRecordData != null){

// 如果hikv里面已经存在播放记录了，需要将cb中旧的播放记录移到原有hikv后面, 并将cb中旧的播放记录用切分好的新的播放记录覆盖

// 日志****** couchbase中超过阈值的单个用户的uid, 第一条记录，记录总共条数，最后一条记录

val parseResultLastTimedRecordIndex = parseResult.getTimedRecordCount - 1

if (parseResultLastTimedRecordIndex != -1){

val parseResultLastTimedRecord = parseResult.getTimedRecord(parseResultLastTimedRecordIndex)

val parseResultLastRecordEntryIndex = parseResultLastTimedRecord.getRecordCount - 1

if (parseResultLastRecordEntryIndex != -1){

val parseResultLastRecordEntry = parseResultLastTimedRecord.getRecord(parseResultLastRecordEntryIndex)

println("0000 couchbase beyond threshold all records uid: " + uid)

println("0000 couchbase beyond threshold all records first record: " + parseResult.getTimedRecord(0).getRecord(0))

println("0000 couchbase beyond threshold all records total count: " + calTotalCount(parseResult))

println("0000 couchbase beyond threshold all records last record: " + parseResultLastRecordEntry)

}

}

// 日志******

// 日志****** couchbase中超过阈值的单个用户的uid,截取的较旧的播放记录的第一条记录，较旧记录的总共条数，最后一条记录

val cbUserPlayRecordsLastTimedRecordIndex = oldCbCutEntity.getCbUserPlayRecords.getTimedRecordCount - 1

if (cbUserPlayRecordsLastTimedRecordIndex != -1){

val cbUserPlayRecordsLastTimedRecord = oldCbCutEntity.getCbUserPlayRecords.getTimedRecord(cbUserPlayRecordsLastTimedRecordIndex)

val cbUserPlayRecordsLastRecordEntryIndex = cbUserPlayRecordsLastTimedRecord.getRecordCount - 1

if (cbUserPlayRecordsLastRecordEntryIndex != -1){

val cbUserPlayRecordsLastRecordEntry = cbUserPlayRecordsLastTimedRecord.getRecord(cbUserPlayRecordsLastRecordEntryIndex)

println("1111 couchbase beyond threshold cut old records uid: " + uid)

println("1111 couchbase beyond threshold cut old records first record: " + oldCbCutEntity.getCbUserPlayRecords.getTimedRecord(0).getRecord(0))

println("1111 couchbase beyond threshold cut old records break record: " + oldCbCutEntity.getBreakAlbumId)

println("1111 couchbase beyond threshold cut old records total count: " + calTotalCount(oldCbCutEntity.getCbUserPlayRecords.build()))

println("1111 couchbase beyond threshold cut old records last record: " + cbUserPlayRecordsLastRecordEntry)

}

}

// 日志******

// 日志****** couchbase中超过阈值的单个用户的uid, 截取的较新的播放记录的第一条记录，较新记录的总共条数，最后一条记录

val newCbUserPlayRecordsLastTimedRecordIndex = newCbUserPlayRecords.getTimedRecordCount - 1

if (newCbUserPlayRecordsLastTimedRecordIndex != -1){

val newCbUserPlayRecordsLastTimedRecord = newCbUserPlayRecords.getTimedRecord(newCbUserPlayRecordsLastTimedRecordIndex)

val newCbUserPlayRecordsLastRecordEntryIndex = newCbUserPlayRecordsLastTimedRecord.getRecordCount - 1

if (newCbUserPlayRecordsLastRecordEntryIndex != -1){

val newCbUserPlayRecordsLastRecordEntry = newCbUserPlayRecordsLastTimedRecord.getRecord(newCbUserPlayRecordsLastRecordEntryIndex)

println("2222 couchbase beyond threshold cut new records uid: " + uid)

println("2222 couchbase beyond threshold cut new records first record: " + newCbUserPlayRecords.getTimedRecord(0).getRecord(0))

println("2222 couchbase beyond threshold cut new records total count: " + calTotalCount(newCbUserPlayRecords.build()))

println("2222 couchbase beyond threshold cut new records last record: " + newCbUserPlayRecordsLastRecordEntry)

}

}

// 日志******

// hikv中已经存在的记录，之前插入的

val playRecord = playRecordData.getPlayRecord

// 日志****** hikv中已经存在的记录，之前插入的

val playRecordLastTimedRecordIndex = playRecord.getTimedRecordCount - 1

if (playRecordLastTimedRecordIndex != -1){

val playRecordLastTimedRecord = playRecord.getTimedRecord(playRecordLastTimedRecordIndex)

val playRecordLastRecordEntryIndex = playRecordLastTimedRecord.getRecordCount - 1

if (playRecordLastRecordEntryIndex != -1){

val playRecordLastRecordEntry = playRecordLastTimedRecord.getRecord(playRecordLastRecordEntryIndex)

println("3333 hikv already exist uid: " + uid)

println("3333 hikv already exist first record: " + playRecord.getTimedRecord(0).getRecord(0))

println("3333 hikv already exist total count: " + calTotalCount(playRecord.build()))

println("3333 hikv already exist last record: " + playRecordLastRecordEntry)

}

}

// 日志******

val beforePlayRecordCount = calTotalCount(playRecord.build())

// 将couchbase旧的播放记录加入到hikv后面形成新的记录 4444 = 3333 + 1111

val newPlayRecord = playRecord.addAllTimedRecord(oldCbCutEntity.getCbUserPlayRecords.getTimedRecordList)

// 日志******

val newPlayRecordLastTimedRecordIndex = newPlayRecord.getTimedRecordCount - 1

if (newPlayRecordLastTimedRecordIndex != -1){

val newPlayRecordLastTimedRecord = newPlayRecord.getTimedRecord(newPlayRecordLastTimedRecordIndex)

val newPlayRecordLastRecordEntryIndex = newPlayRecordLastTimedRecord.getRecordCount - 1

if (newPlayRecordLastRecordEntryIndex != -1){

val newPlayRecordLastRecordEntry = newPlayRecordLastTimedRecord.getRecord(newPlayRecordLastRecordEntryIndex)

println("4444 hikv ideal uid: " + uid)

println("4444 hikv ideal first record: " + newPlayRecord.getTimedRecord(0).getRecord(0))

println("4444 hikv ideal total count: " + calTotalCount(newPlayRecord.build()))

println("4444 hikv ideal last record: " + newPlayRecordLastRecordEntry)

}

}

// 日志******

// 将hikv中新增的旧的播放记录重新写入到hikv

PlayRecordDataDao.getInstance().set(new PlayRecordData(uid, newPlayRecord)).toCompletableFuture.get // 阻塞写入

// hikv写入成功后，再写入couchbase

/**

* cb hikv

* 成功 失败 不存在 这种情况会导致播放记录减少，因为截取的新的播放记录写入了cb, 但现在先写入hikv, 再写入cb，自然避免了这种情况

* 失败 成功 存在 这种情况会有重复的播放记录，hikv里面有旧的播放记录，cb里面有旧的播放记录和新的播放记录

* 成功 成功 存在 理想情况

* 失败 失败 存在 没有转移成功，下次kafka里面还会发送这个uid过来

*/

writeResult.put(new CbCasEntity(uid, cas), newCbUserPlayRecords.build().toByteArray)

}else{

// 如果hikv里面本来就没有播放记录, 切分，将旧的播放记录放到hikv中，将新的播放记录覆盖到cb中

// 日志****** couchbase中超过阈值的单个用户的uid, 第一条记录，记录总共条数，最后一条记录

val parseResultLastTimedRecordIndex = parseResult.getTimedRecordCount - 1

if (parseResultLastTimedRecordIndex != -1){

val parseResultLastTimedRecord = parseResult.getTimedRecord(parseResultLastTimedRecordIndex)

val parseResultLastRecordEntryIndex = parseResultLastTimedRecord.getRecordCount - 1

if (parseResultLastRecordEntryIndex != -1){

val parseResultLastRecordEntry = parseResultLastTimedRecord.getRecord(parseResultLastRecordEntryIndex)

println("5555 couchbase beyond threshold all records uid: " + uid)

println("5555 couchbase beyond threshold all records first record: " + parseResult.getTimedRecord(0).getRecord(0))

println("5555 couchbase beyond threshold all records total count: " + calTotalCount(parseResult))

println("5555 couchbase beyond threshold all records last record: " + parseResultLastRecordEntry)

}

}

// 日志******

// 日志****** couchbase中超过阈值的单个用户的uid,截取的较旧的播放记录的第一条记录，较旧记录的总共条数，最后一条记录

val cbUserPlayRecordsLastTimedRecordIndex = oldCbCutEntity.getCbUserPlayRecords.getTimedRecordCount - 1

if (cbUserPlayRecordsLastTimedRecordIndex != -1){

val cbUserPlayRecordsLastTimedRecord = oldCbCutEntity.getCbUserPlayRecords.getTimedRecord(cbUserPlayRecordsLastTimedRecordIndex)

val cbUserPlayRecordsLastRecordEntryIndex = cbUserPlayRecordsLastTimedRecord.getRecordCount - 1

if (cbUserPlayRecordsLastRecordEntryIndex != -1){

val cbUserPlayRecordsLastRecordEntry = cbUserPlayRecordsLastTimedRecord.getRecord(cbUserPlayRecordsLastRecordEntryIndex)

println("6666 couchbase beyond threshold cut old records uid: " + uid)

println("6666 couchbase beyond threshold cut old records first record: " + oldCbCutEntity.getCbUserPlayRecords.getTimedRecord(0).getRecord(0))

println("6666 couchbase beyond threshold cut old records total count: " + calTotalCount(oldCbCutEntity.getCbUserPlayRecords.build()))

println("6666 couchbase beyond threshold cut old records last record: " + cbUserPlayRecordsLastRecordEntry)

}

}

// 日志******

// 日志****** couchbase中超过阈值的单个用户的uid, 截取的较新的播放记录的第一条记录，较新记录的总共条数，最后一条记录

val newCbUserPlayRecordsLastTimedRecordIndex = newCbUserPlayRecords.getTimedRecordCount - 1

if (newCbUserPlayRecordsLastTimedRecordIndex != -1){

val newCbUserPlayRecordsLastTimedRecord = newCbUserPlayRecords.getTimedRecord(newCbUserPlayRecordsLastTimedRecordIndex)

val newCbUserPlayRecordsLastRecordEntryIndex = newCbUserPlayRecordsLastTimedRecord.getRecordCount - 1

if (newCbUserPlayRecordsLastRecordEntryIndex != -1){

val newCbUserPlayRecordsLastRecordEntry = newCbUserPlayRecordsLastTimedRecord.getRecord(newCbUserPlayRecordsLastRecordEntryIndex)

println("7777 couchbase beyond threshold cut new records uid: " + uid)

println("7777 couchbase beyond threshold cut new records first record: " + newCbUserPlayRecords.getTimedRecord(0).getRecord(0))

println("7777 couchbase beyond threshold cut new records total count: " + calTotalCount(newCbUserPlayRecords.build()))

println("7777 couchbase beyond threshold cut new records last record: " + newCbUserPlayRecordsLastRecordEntry)

}

}

// 日志******

PlayRecordDataDao.getInstance().set(new PlayRecordData(uid, oldCbCutEntity.getCbUserPlayRecords)).toCompletableFuture.get

writeResult.put(new CbCasEntity(uid, cas), newCbUserPlayRecords.build().toByteArray)

}

}catch {

case e: Exception => e.printStackTrace()

}

}

val errorWriteResult = collection.mutable.Set[String]()

CbManager.getInstance().writeCasAsync("test_playback_record_couchbase", writeResult.asJava, 186, 16070400, errorWriteResult.asJava)

println("writeResult: " + writeResult.keySet.map(item => item.getId)) // 这些key说明hikv确实是插入成功了, 可以查看写入cb的key有哪些

val writeSuccess = writeResult.keySet.filterNot(item => errorReadResult.contains(item.getId))

println("writeSuccess: " + writeSuccess.map(item => item.getId))

}

}

}

ssc.start()

ssc.awaitTermination()

}

def calTotalCount(parseResult: PlayRecord.UserPlayRecords): Int = {

/*

计算从couchbase解析出来的每个用户的播放记录长度

*/

var recordTotalCount = 0

for (timeRecordItem

recordTotalCount += timeRecordItem.getRecordCount

}

recordTotalCount

}

def calTransferCount(parseResult: PlayRecord.UserPlayRecords, cbSingleUerThreshold: Int): Int = {

/*

计算超过阈值的单个用户从couchbase转移到hikv的播放记录的长度

e.g. : 10000(couchbase中存在的量) - 1000(单个用户的阈值数) = 9000

*/

var recordTotalCount = 0

for (timeRecordItem

recordTotalCount += timeRecordItem.getRecordCount

}

recordTotalCount - cbSingleUerThreshold

}

def cutOldPlayRecordsFromCb(transferCount: Int, parseResult: PlayRecord.UserPlayRecords): CbCutEntity = {

/*

将couchbase中单个用户的播放记录进行切分，得到旧的播放记录

*/

var breakCount = 0

var recordEntryCount = 0 // 看break的那个recordEntry是到达第几个以后退出

var timeRecordBreakCount = 0 // 之前转移到hikv的最后的timeRecord的点

var breakFlag = false

var breakAlbumId = 0L

var userPlayRecords = PlayRecord.UserPlayRecords.newBuilder()

breakable{for (timedRecordItem

var timeRecord = PlayRecord.TimedRecord.newBuilder()

timeRecord.setRecordTime(timedRecordItem.getRecordTime)

var innerRecordEntryCount = 0 // 每一次都初始化为0

breakable {for (recordEntryItem

var recordEntry = PlayRecord.RecordEntry.newBuilder()

recordEntry.setAlbumId(recordEntryItem.getAlbumId)

recordEntry.setMaxTvId(recordEntryItem.getMaxTvId)

recordEntry.setMaxPlayPosition(recordEntryItem.getMaxPlayPosition)

if (breakCount == transferCount){

recordEntryCount = innerRecordEntryCount

breakAlbumId = recordEntryItem.getAlbumId

breakFlag = true

break

}

timeRecord.addRecord(recordEntry)

breakCount += 1

innerRecordEntryCount += 1

}}

userPlayRecords.addTimedRecord(timeRecord)

if (breakFlag){

break

}

timeRecordBreakCount += 1

}}

new CbCutEntity(timeRecordBreakCount, recordEntryCount, breakAlbumId, userPlayRecords)

}

def cutNewPlayRecordsFromCb(timeRecordBreakCount: Int, recordEntryCount: Int, parseResult: PlayRecord.UserPlayRecords): PlayRecord.UserPlayRecords.Builder = {

var cbUserPlayRecords = PlayRecord.UserPlayRecords.newBuilder()

for (trIndex

var timedRecordItem = parseResult.getTimedRecord(trIndex)

var cbTimeRecord = PlayRecord.TimedRecord.newBuilder()

cbTimeRecord.setRecordTime(timedRecordItem.getRecordTime)

if (trIndex == timeRecordBreakCount) {

for (rIndex

var recordEntryItem = timedRecordItem.getRecord(rIndex)

var cbRecordEntry = PlayRecord.RecordEntry.newBuilder()

cbRecordEntry.setAlbumId(recordEntryItem.getAlbumId)

cbRecordEntry.setMaxTvId(recordEntryItem.getMaxTvId)

cbRecordEntry.setMaxPlayPosition(recordEntryItem.getMaxPlayPosition)

cbTimeRecord.addRecord(cbRecordEntry)

}

}else for (recordEntryItem

var cbRecordEntry = PlayRecord.RecordEntry.newBuilder()

cbRecordEntry.setAlbumId(recordEntryItem.getAlbumId)

cbRecordEntry.setMaxTvId(recordEntryItem.getMaxTvId)

cbRecordEntry.setMaxPlayPosition(recordEntryItem.getMaxPlayPosition)

cbTimeRecord.addRecord(cbRecordEntry)

}

cbUserPlayRecords.addTimedRecord(cbTimeRecord)

}

cbUserPlayRecords

}

def compareOldNew(beforePlayRecordCount: Int, newPlayRecord: PlayRecord.UserPlayRecords.Builder): Boolean = {

var newRecordTotalCount = 0

for (timedRecordItem

newRecordTotalCount += timedRecordItem.getRecordCount

}

newRecordTotalCount > beforePlayRecordCount

}

}