什么是数据丢失

两个线程基于同一个查询结果进行修改,后修改的人会将先修改人的修改覆盖掉. 
让我们先来看这么个小案例： 
我们给游戏充值100，支付成功后，银行会向游戏服务器发送支付成功信息，有一个订单支付信息表（order）和一个账户信息表（account），首先要去order表中查询该订单支付状态state（select state from order where id=1）如果是0未支付，接下来要做的就是更新state为已支付，并且向你的账户account中加上100. 但是网络是有延迟的银行为了确保信息能送达可能会再次向服务器发送支付成功信息。这时第二次请求也是先去查询state，若此次请求和第一次正好是并发执行，他查询到的state也是0，所以也会给你的account加100块钱。也就是后来的修改覆盖掉了前边的修改，这就是数据丢失。 

操作流程如下： 

怎么防止数据丢失

先来看一下数据库中的锁机制: 
共享锁:在非Serializable隔离级别做查询不加任何锁,而在Serializable隔离级别下做的查询加共享锁, 
共享锁的特点:共享锁和共享锁可以共存,但是共享锁和排他锁不能共存 
排他锁:在所有隔离级别下进行增删改的操作都会加排他锁, 
排他锁的特点:和任意其他锁都不能共存

悲观锁：

悲观锁悲观锁悲观的认为每一次操作都会造成更新丢失问题,在每次查询时就加上排他锁。 
select state from order for update 
这样第二个请求在查询时发现已经有个排他锁，就在那等着，等第一个事务操作完成后才轮到他，不过此时state已经变成1已支付，这样就避免再次加100块。

乐观锁:

乐观锁会乐观的认为每次查询都不会造成更新丢失.利用一个版本字段进行控制。 
为order表加一个字段version 
第一次请求查询出未支付后，在加100块钱之前,将更新state操作修改为： 
update order set state=1 and version =version+1 where id=1 and version=0 
这样第二个请求也执行这句是version就不是0了，执行失败。从而也防止了第一次数据的丢失。

两种方案对比

• 查询非常多,修改非常少,使用乐观锁（悲观查询都加排他锁 效率会降低）

• 修改非常多,查询非常少,使用悲观锁（多管理员修改时，某种一个可能一直等待，所以用悲观）转载

一、Kafka优化总结

翻译原文如下：

https://www.infoq.com/articles/apache-kafka-best-practices-to-optimize-your-deployment

1. 设置日志配置参数以使日志易于管理

kafka 日志文档 https://kafka.apache.org/documentation/#log

kafka 压缩基础知识  https://kafka.apache.org/documentation/#design_compactionbasics

 

2. 了解 kafka 的 (低) 硬件需求

CPU ：除非需要SSL和日志压缩，否则Kafka不需要强大的CPU。使用的内核越多，并行性能越好，大多数情况下压缩，压缩也不会产生影响，应该使用LZ4 编码器来提供最佳性能。

RAM：对特别重的负载生产负载，使用32GB一上的机器，将提高客户端吞吐量

磁盘：如果在RAIO设置中使用多个驱动器，就该Kafka大显身手。由于Kafka的顺序磁盘I/O范式，所有SSD不会提供太多的优势，不应该使用NAS

网络和文件系统：建议使用XFS，如果条件允许，还可以将集群放在单个数据中心。应尽量提供更多的网络带宽。

• Apache Kafka 的基准测试：每秒 200 万次写 (在 3 台廉价的机器上)

  https://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines

• 在 AWS 上的 Apache Kafka 负载测试

  https://grey-boundary.io/load-testing-apache-kafka-on-aws/

• 性能测试

  https://cwiki.apache.org/confluence/display/KAFKA/Performance+testing

3. 充分利用 Apache ZooKeeper

Zookeeper 节点的数量最大应该是五个。一个节点适合于开发环境，三个节点对于大多数产品Kafka集群足够了，虽然一个大型Kafka集群可能需要5个节点来减少延迟，必须考虑节点负载，近期版本的Kafka对Zookeeper的负载要低很多，早期版本使用Zookeeper来存储消费者偏移量。

为Zookeeper 提供最强大的网络带宽。使用最好的磁盘，分别存储日志、隔离Zookeeper进程、禁用交换，这些也会减少延迟。

4. 以正确的方式设置复制和冗余

5. 注意主题配置

6. 使用并行处理

7. 带着安全性思维配置和隔离 Kafka

8. 通过提高限制避免停机 Ulimit

1. 1、创建一个新的文件：/etc/security/limits.d/nofile.conf

2. 2、输入内容：

   • soft nofile 128000
     
     hard nofile 128000

      

3. 3、重新启动系统或重新登录。

4. 4、通过以下命令来验证

5. ulimit  -a 

    

9. 保持低网络延迟

10. 利用有效的监控和警报

通过 Instaclustr 控制台中显示的 Kafka 监控图示例：监视系统指标 (如网络吞吐量、打开的文件句柄、内存、负载、磁盘使用情况和其他因素) 是必不可少的，同时还要密切关注 JVM 统计数据，包括 GC 暂停和堆使用情况。仪表板和历史回溯工具能够加速调试过程，可以提供大量的价值。与此同时，应该配置 Nagios 或 PagerDuty 等警报系统，以便在出现延迟峰值或磁盘空间不足等症状时发出警告

二、Kafka数据丢失解决方案

producer 数据不丢失：

1. 同步模式：配置=1 （只有Leader收到，-1 所有副本成功，0 不等待）Leader Partition挂了，数据就会丢失

解决：设置 -1 保证produce 写入所有副本算成功 producer.type = sync request.required.acks=-1

2. 异步模式，当缓冲区满了，如果配置为0（没有收到确认，一满就丢弃），数据立刻丢弃

解决：不限制阻塞超时时间。就是一满生产者就阻塞

producer.type = async

request.required.acks=1

queue.buffering.max.ms=5000

queue.buffering.max.messages=10000

queue.enqueue.timeout.ms = -1

batch.num.messages=200

Customer 不丢失数据

在获取kafka的消息后正准备入库（未入库），但是消费者挂了，那么如果让kafka自动去维护offset ，它就会认为这条数据已经被消费了，那么会造成数据丢失。

解决：使用kafka高级API，自己手动维护偏移量，当数据入库之后进行偏移量的更新（适用于基本数据源）

流式计算。高级数据源以kafka为例，由2种方式：receiver (开启WAL，失败可恢复) director （checkpoint保证）

流处理中的几种可靠性语义：

1. at most once  每条数据最多被处理一次（0次或1次），会出现数据丢失的问题

2. at least once 每条数据最少被处理一次（1次或更多），这个不会出现数据丢失，但是会出现数据重复

3. exactly once 每种数据只会被处理一次，没有数据丢失，没有数据重复，这种语义是大家最想实现的，也是最难实现的

但是开启WAL后，依旧存在数据丢失问题，原因是任务中断时receiver 也被强行终止了，将会造成数据丢失

在Streaming程序的最后添加代码，只有在确认所有receiver都关闭的情况下才终止程序

 

 

三、Kafka特点、优点、应用场景

1、kafka 特点

分布式：1.多分区 2. 多副本  3. 多订阅者（订阅者数量小于等于partition数量）4. 基于Zookeeper调度

高性能：1. 高吞吐量 2. 低延迟  3. 高并发 4. 时间复杂度为O（1）

持久性与扩展性：1. 数据可持久化  2. 容错性（多副本和Consumer Group 支持了容错性） 3.支持在线水平扩展（Broker 有一个或多个Partition ，Topic 有一个或多个Partition，Consumer Group 变化，我们增加了新的机器，就可以放新的topic和新的Partition）4. 消息自动平衡（消息在服务端进行平衡，consumer在消费的时候进行平衡）

Kafka 高级特性--消息事务

数据传输的事务定义：

最多一次：消息不会被重复发送、最多被传输一次、但也有可能一次不传输

最少一次：消息不会被漏发送，最少被传输一次，但也有可能重复传输（消费端需要判断处理）

精确的一次（Exactly once）：不会漏传输也不会重复传输，每个消息都传输被一次而且仅仅传输一次

事务保证

Produce幂等处理、多分区原子写入（事务要保证kafka下每一分区的原子写入，原子是一个读取--处理--写入操作）

事务保证--避免僵尸实例

每个事务Producer 分配一个transactional.id ,在进程重新启动时能够识别相同的Producer实例

kafka 增加了一个与transactional.id 相关的epoch, 存储每个transactional.id 内部元数据

一旦epoch 被触发，任何具有相同的transactional.id 和更旧的epoch的Producer 被视为僵尸，Kafka会拒绝来自这些Procedure的后续事务写入

kafka高性能--零拷贝

网络传输持久性日志块（消息）

零拷贝建立在：Java Nio channel.transforTo()方法  实际是调用  Linux sendfile 系统调用

文件传输到网络的公共数据路径

1. 操作系统将数据从磁盘读入到内核空间的页缓存

2. 应用程序将数据从内核空间读入到用户空间缓存中

3. 应用程序将数据写回到内核空间到socket 缓存中

4. 操作系统将数据从socket缓冲区复制到网卡缓冲区，一便将数据经网络发出

零拷贝过程：（内核空间和用户空间零拷贝）

1.操作系统将数据从磁盘读入到内核空间的页缓存

2.将数据的位置和长度的信息的描述符增加至内核空间（socket缓存区）

3.操作系统将数据从内核拷贝到网卡缓冲区，以便将数据经网络发出

2、Kafka优点

高吞吐量、低延迟、高并发、高性能的消息中间件。Kafka 集群甚至可以做到每秒几十万、上百万的超高并发写入

页面缓存技术+磁盘顺序写+零拷贝技术

3、Kafka 应用场景

1. 消息对列

 2. 行为跟踪（用户的行为跟踪直接发送到topic ，时时处理或离线数据存储）

3. 元数据监控（操作信息运维数据监控）

4. 日志收集

5. 流处理（收集上游流数据，经过处理，发布到新的topic 中，进行处理，中间可以进行实时计算、数据处理）

6. 事件源（将状态转移作为记录，按照时间顺序的序列，我们可以回溯整个时间的变更）

7. 持久性日志（commit log ，就是说在节点之后进行一个持久性日志记录，日志可以在节点间备份一份数据，并且给故障节点间恢复提供一种机制）

 

 

 

 

 

 

 

 

 

 

 

 

 

 

1、问题描述：

一般在登录成功的时候需要把用户信息，菜单信息放置vuex中，作为全局的共享数据。但是在页面刷新的时候vuex里的数据会重新初始化，导致数据丢失。因为vuex里的数据是保存在运行内存中的，当页面刷新时，页面会重新加载vue实例，vuex里面的数据就会被重新赋值。

2、解决思路：

办法一：将vuex中的数据直接保存到浏览器缓存中（sessionStorage、localStorage、cookie）

办法二：在页面刷新的时候再次请求远程数据，使之动态更新vuex数据

办法三：在父页面向后台请求远程数据，并且在页面刷新前将vuex的数据先保存至sessionStorage（以防请求数据量过大页面加载时拿不到返回的数据）

分析：

办法一的缺点是不安全，不适用大数据量的存储；

办法二适用于少量的数据，并且不会出现网络延迟；

办法三是要讲的重点，办法二和办法一一起使用。

3、解决过程：@1

3.1、选择合适的浏览器存储

localStorage -- 是永久存储在本地，除非你主动去删除;

sessionStorage -- 是存储到当前页面关闭为止，和其他tab页没关联;

cookie -- 则根据你设置的有效时间来存储，但缺点是不能储存大数据且不易读取，会和后台进行交互。

本方法选择的是sessionStorage，选择的原因是由于vue是单页面应用，操作都是在一个页面跳转路由，另一个原因是sessionStorage可以保证打开页面时sessionStorage的数据为空，而如果是localStorage则会读取上一次打开页面的数据。

3.2、解决方案

由于state里的数据是响应式，所以sessionStorage存储也要跟随变化，而且只能通过mutations来改变state中的值。

首先在用户登录成功之后，然后把用户信息、菜单信息通过actions分发保存至vuex中。然后在菜单页面计算vuex中state的菜单数据，将数据解析组装成前端组件所需的格式，然后渲染组件，生成菜单树。如果页面没有刷新，则一切正常。

登录成功后将用户和菜单数据同步至vuex

在菜单页面监听vuex中菜单数据

页面刷新的解决方案：

页面刷新的时候异步请求后台数据，然后动态更新vuex中的数据，其中会有一种情况就是，网络延迟、数据量大的问题。此时还没等vuex获取到后台返回的数据，页面就已经加载完成了，这样就会造成数据丢失。所以该解决方案就是，监听浏览器刷新前事件，在浏览器刷新之前就把vuex里的数据保存至sessionStorage中，刷新成功后如果异步请求的数据还没返回则直接获取sessionStorage里的数据，否则获取vuex里的数据。（只有刷新后还没取到后台数据，才会从sessionStorage里取。确保数据的安全性，就算获取sessionStorage里的数据也是安全的，因为每次刷新都会重新赋值，不必担心数据被篡改问题，其次就是对sessionStorage里的数据做了加密操作）

在父页面向后台请求数据，并且监听浏览器刷新前事件，将vuex数据保存至sessionStorage

在父页面向后台请求数据，将返回的数据分发至vuex

3、解决过程：@2 vuex-persistedstate

原理：

将vuex的state存在localStorage或sessionStorage或cookie中一份

刷新页面的一瞬间，vuex数据消失，vuex会去sessionStorage中拿回数据，变相的实现了数据刷新不丢失~

使用方法：

安装：npm install vuex-persistedstate --save

在store下的index.js中，引入并配置

import createPersistedState from "vuex-persistedstate"

const store = new Vuex.Store({
  // ...
  plugins: [createPersistedState()]
})

此时可以选择数据存储的位置，可以是localStorage/sessionStorage/cookie，此处以存储到sessionStorage为例，配置如下：

import createPersistedState from "vuex-persistedstate"
const store = new Vuex.Store({
  // ...
  plugins: [createPersistedState({
      storage: window.sessionStorage
  })]
})

存储指定state：

vuex-persistedstate默认持久化所有state，指定需要持久化的state,配置如下：

import createPersistedState from "vuex-persistedstate"
const store = new Vuex.Store({
  // ...
  plugins: [createPersistedState({
      storage: window.sessionStorage,
      reducer(val) {
          return {
          // 只储存state中的user
          user: val.user
        }
     }
  })]

此刻的val 对应store/modules文件夹下几个js文件存储的内容，也就是stor/index中import的几个模块，可以自己打印看看。希望哪一部分的数据持久存储，将数据的名字在此配置就可以，目前我只想持久化存储user模块的数据。

注意：如果此刻想配置多个选项，将plugins写成一个一维数组，不然会报错。

import createPersistedState from "vuex-persistedstate"
import createLogger from 'vuex/dist/logger'
// 判断环境 vuex提示生产环境中不使用
const debug = process.env.NODE_ENV !== 'production'
const createPersisted = createPersistedState({
  storage: window.sessionStorage
})
export default new Vuex.Store({
 // ...
  plugins: debug ? [createLogger(), createPersisted] : [createPersisted]
})

3、解决过程：@3 vuex-along

将state里的数据保存一份到本地存储(localStorage、sessionStorage、cookie）

//App.vue

<script>
    export default{
        created() {
            //在页面加载时读取sessionStorage里的状态信息
             if (sessionStorage.getItem("store") ) {
                this.$store.replaceState(Object.assign({}, this.$store.state,JSON.parse(sessionStorage.getItem("store"))))
            }

            //在页面刷新时将vuex里的信息保存到sessionStorage里
            window.addEventListener("beforeunload",()=>{
                sessionStorage.setItem("store",JSON.stringify(this.$store.state))
            })
        }
    }
</script>

2.使用第三方插件如vuex-along（本质上还是利用session等缓存到本地）

import Vue from 'vue'
import Vuex from 'vuex'
import VueXAlong from 'vuex-along'
Vue.use(Vuex)
let productData = require('@/assets/api/list.json')

const moduleA = {
    state: {
        a1: 'a1',
        a2: 'a2',
    }
}
export default new Vuex.Store({
    state: {
        productList: [],//列表数据
        cartList: [],//购物车数据
    },
    mutations: {
        //产品列表
        setProductList(state, data){
            state.productList = data
        },
        //加入购物车
        setCartList(state, id){
            let  cartIndex = state.cartList.findIndex(item => item.id === id)
            if(cartIndex < 0){
                state.cartList.push({id, count: 1})
            }else{
                state.cartList[cartIndex].count++
            }
        },
        //删除购物车商品
        deleteCartList(state, id){
            let  cartIndex = state.cartList.findIndex(item => item.id === id)
            state.cartList.splice(cartIndex, 1)
        },
        //编辑购物车商品数量
        editCartList(state, data){
            let cartIndex = state.cartList.findIndex(item => item.id === data.id)
            state.cartList[cartIndex].count = data.count
        },
        clearCart(state){
            state.cartList = []
        },
    },
    actions: {
        getProductList(context){
            //模拟ajax请求,将返回的信息直接存储在store
            setTimeout(()=>{
                context.commit('setProductList', productData)
            }, 2000)
        },
        buy(context){
            //模拟ajax请求通过返回promise对象将结果返回给操作提交的地方
            return new Promise((resolve, reject) => {
                setTimeout(()=>{
                    context.commit('clearCart')
                    resolve({msg:'下单成功', code: 200})
                    //reject({message: '提交失败', code: 300})
                }, 1000)
            })
        }
    },
    modules: {
        ma: moduleA
    },
    //缓存state的数据到storage
    plugins: [VueXAlong()],
    //全量的参数配置(sessionStorage 数据恢复优先级高于 localStorage)
    /* plugins: [VueXAlong({
        // 设置保存的集合名字，避免同站点下的多项目数据冲突
        name: 'my-app-VueXAlong',
        //过滤ma的数据将其他的数据存入localStorage
        local: {
            list: ['ma'],//需要监听的属性名或者模块名
            isFilter: true,//是否过滤而非保存
        },
        //保存ma的a1数据到sessionStorage
        session: {
            list: ['ma.a1'],
            isFilter: false,
        },
        //仅使用sessionStorage保存
        //justSession: true,
    })] */
})