[Java中存储不同类型数据的解决方案]的搜索结果

...之后发生错误 比如，类型错误，比如对 String 使用了 Hash 的命令，这是一种运行时错误。 最后我们发现 set k1 1 的命令是成功的，也就是在这种发生了运行时异常的情况下， 只有错误的命令没有被执行，但是其他命令没有受到影响。 这个显然不符合我们对原子性的定义，也就是我们没办法用 Redis 的这种事务机制来实现原子性，保证数据的一致。 3、Lua脚本 Lua/ˈluə/是一种轻量级脚本语言，它是用 C 语言编写的，跟数据的存储过程有点类似。 使用 Lua 脚本来执行 Redis 命令的好处: 1、一次发送多个命令，减少网络开销。 2、Redis 会将整个脚本作为一个整体执行，不会被其他请求打断，保持原子性。 3、对于复杂的组合命令，我们可以放在文件中，可以实现程序之间的命令集复用。 3.1 在Redis中调用Lua脚本 使用 eval /ɪ’væl/ 方法，语法格式: redis> eval lua-script key-num [key1 key2 key3 ....] [value1 value2 value3 ....] eval代表执行Lua语言的命令。 lua-script代表Lua语言脚本内容。 key-num表示参数中有多少个key，需要注意的是Redis中key是从1开始的，如果没有key的参数，那么写0。 [key1key2key3…]是key作为参数传递给Lua语言，也可以不填，但是需要和key-num的个数对应起来。 [value1 value2 value3 …]这些参数传递给 Lua 语言，它们是可填可不填的。 示例，返回一个字符串，0 个参数: redis> eval "return 'Hello World'" 0 3.2 在Lua脚本中调用Redis命令 使用 redis.call(command, key [param1, param2…])进行操作。语法格式: redis> eval "redis.call('set',KEYS[1],ARGV[1])" 1 lua-key lua-value command是命令，包括set、get、del等。 key是被操作的键。 param1,param2…代表给key的参数。 注意跟 Java 不一样，定义只有形参，调用只有实参。 Lua 是在调用时用 key 表示形参，argv 表示参数值(实参)。 3.2.1 设置键值对 在 Redis 中调用 Lua 脚本执行 Redis 命令 redis> eval "return redis.call('set',KEYS[1],ARGV[1])" 1 gupao 2673 redis> get gupao 以上命令等价于 set gupao 2673。 在 redis-cli 中直接写 Lua 脚本不够方便，也不能实现编辑和复用，通常我们会把脚本放在文件里面，然后执行这个文件。 3.2.2 在 Redis 中调用 Lua 脚本文件中的命令，操作 Redis 创建 Lua 脚本文件: cd /usr/local/soft/redis5.0.5/src vim gupao.lua Lua 脚本内容，先设置，再取值: cd /usr/local/soft/redis5.0.5/src redis-cli --eval gupao.lua 0 得到返回值: root@localhost src] redis-cli --eval gupao.lua 0 "lua666" 3.2.3 案例:对 IP 进行限流 需求：在 X 秒内只能访问 Y 次。 设计思路：用 key 记录 IP，用 value 记录访问次数。 拿到 IP 以后，对 IP+1。如果是第一次访问，对 key 设置过期时间(参数 1)。否则判断次数，超过限定的次数(参数 2)，返回 0。如果没有超过次数则返回 1。超过时间， key 过期之后，可以再次访问。 KEY[1]是 IP， ARGV[1]是过期时间 X，ARGV[2]是限制访问的次数 Y。 -- ip_limit.lua-- IP 限流，对某个 IP 频率进行限制 ，6 秒钟访问 10 次 local num=redis.call('incr',KEYS[1])if tonumber(num)==1 thenredis.call('expire',KEYS[1],ARGV[1])return 1elseif tonumber(num)>tonumber(ARGV[2]) thenreturn 0 elsereturn 1 end 6 秒钟内限制访问 10 次，调用测试(连续调用 10 次): ./redis-cli --eval "ip_limit.lua" app:ip:limit:192.168.8.111 , 6 10 app:ip:limit:192.168.8.111 是 key 值 ，后面是参数值，中间要加上一个空格和一个逗号，再加上一个空格 。 即:./redis-cli –eval [lua 脚本] [key…]空格,空格[args…] 多个参数之间用一个空格分割 。 代码:LuaTest.java 3.2.4 缓存 Lua 脚本 为什么要缓存 在脚本比较长的情况下，如果每次调用脚本都需要把整个脚本传给 Redis 服务端， 会产生比较大的网络开销。为了解决这个问题，Redis 提供了 EVALSHA 命令，允许开发者通过脚本内容的 SHA1 摘要来执行脚本。 如何缓存 Redis 在执行 script load 命令时会计算脚本的 SHA1 摘要并记录在脚本缓存中，执行 EVALSHA 命令时 Redis 会根据提供的摘要从脚本缓存中查找对应的脚本内容，如果找到了则执行脚本，否则会返回错误:“NOSCRIPT No matching script. Please use EVAL.” 127.0.0.1:6379> script load "return 'Hello World'" "470877a599ac74fbfda41caa908de682c5fc7d4b"127.0.0.1:6379> evalsha "470877a599ac74fbfda41caa908de682c5fc7d4b" 0 "Hello World" 3.2.5 自乘案例 Redis 有 incrby 这样的自增命令，但是没有自乘，比如乘以 3，乘以 5。我们可以写一个自乘的运算，让它乘以后面的参数： local curVal = redis.call("get", KEYS[1]) if curVal == false thencurVal = 0 elsecurVal = tonumber(curVal)endcurVal = curVal tonumber(ARGV[1]) redis.call("set", KEYS[1], curVal) return curVal 把这个脚本变成单行，语句之间使用分号隔开 local curVal = redis.call("get", KEYS[1]); if curVal == false then curVal = 0 else curVal = tonumber(curVal) end; curVal = curVal tonumber(ARGV[1]); redis.call("set", KEYS[1], curVal); return curVal script load ‘命令’ 127.0.0.1:6379> script load 'local curVal = redis.call("get", KEYS[1]); if curVal == false then curVal = 0 else curVal = tonumber(curVal) end; curVal = curVal tonumber(ARGV[1]); redis.call("set", KEYS[1], curVal); return curVal' "be4f93d8a5379e5e5b768a74e77c8a4eb0434441" 调用: 127.0.0.1:6379> set num 2OK127.0.0.1:6379> evalsha be4f93d8a5379e5e5b768a74e77c8a4eb0434441 1 num 6 (integer) 12 3.2.6 脚本超时 Redis 的指令执行本身是单线程的，这个线程还要执行客户端的 Lua 脚本，如果 Lua 脚本执行超时或者陷入了死循环，是不是没有办法为客户端提供服务了呢? eval 'while(true) do end' 0 为了防止某个脚本执行时间过长导致 Redis 无法提供服务，Redis 提供了 lua-time-limit 参数限制脚本的最长运行时间，默认为 5 秒钟。 lua-time-limit 5000(redis.conf 配置文件中) 当脚本运行时间超过这一限制后，Redis 将开始接受其他命令但不会执行(以确保脚本的原子性，因为此时脚本并没有被终止)，而是会返回“BUSY”错误。 Redis 提供了一个 script kill 的命令来中止脚本的执行。新开一个客户端: script kill 如果当前执行的 Lua 脚本对 Redis 的数据进行了修改(SET、DEL 等)，那么通过 script kill 命令是不能终止脚本运行的。 127.0.0.1:6379> eval "redis.call('set','gupao','666') while true do end" 0 因为要保证脚本运行的原子性，如果脚本执行了一部分终止，那就违背了脚本原子性的要求。最终要保证脚本要么都执行，要么都不执行。 127.0.0.1:6379> script kill(error) UNKILLABLE Sorry the script already executed write commands against the dataset. You can either wait the scripttermination or kill the server in a hard way using the SHUTDOWN NOSAVE command. 遇到这种情况，只能通过 shutdown nosave 命令来强行终止 redis。 shutdown nosave 和 shutdown 的区别在于 shutdown nosave 不会进行持久化操作，意味着发生在上一次快照后的数据库修改都会丢失。 4、Redis 为什么这么快? 4.1 Redis到底有多快？ 根据官方的数据，Redis 的 QPS 可以达到 10 万左右(每秒请求数)。 4.2 Redis为什么这么快? 总结:1)纯内存结构、2)单线程、3)多路复用 4.2.1 内存 KV 结构的内存数据库，时间复杂度 O(1)。 第二个，要实现这么高的并发性能，是不是要创建非常多的线程? 恰恰相反，Redis 是单线程的。 4.2.2 单线程 单线程有什么好处呢? 1、没有创建线程、销毁线程带来的消耗 2、避免了上线文切换导致的 CPU 消耗 3、避免了线程之间带来的竞争问题，例如加锁释放锁死锁等等 4.2.3 异步非阻塞 异步非阻塞 I/O，多路复用处理并发连接。 4.3 Redis为什么是单线程的? 不是白白浪费了 CPU 的资源吗? 因为单线程已经够用了，CPU 不是 redis 的瓶颈。Redis 的瓶颈最有可能是机器内存或者网络带宽。既然单线程容易实现，而且 CPU 不会成为瓶颈，那就顺理成章地采用单线程的方案了。 4.4 单线程为什么这么快? 因为 Redis 是基于内存的操作，我们先从内存开始说起。 4.4.1 虚拟存储器(虚拟内存 Vitual Memory) 名词解释:主存:内存;辅存:磁盘(硬盘) 计算机主存(内存)可看作一个由 M 个连续的字节大小的单元组成的数组，每个字节有一个唯一的地址，这个地址叫做物理地址(PA)。早期的计算机中，如果 CPU 需要内存，使用物理寻址，直接访问主存储器。 这种方式有几个弊端: 1、在多用户多任务操作系统中，所有的进程共享主存，如果每个进程都独占一块物理地址空间，主存很快就会被用完。我们希望在不同的时刻，不同的进程可以共用同一块物理地址空间。 2、如果所有进程都是直接访问物理内存，那么一个进程就可以修改其他进程的内存数据，导致物理地址空间被破坏，程序运行就会出现异常。 为了解决这些问题，我们就想了一个办法，在 CPU 和主存之间增加一个中间层。CPU 不再使用物理地址访问，而是访问一个虚拟地址，由这个中间层把地址转换成物理地址，最终获得数据。这个中间层就叫做虚拟存储器(Virtual Memory)。 具体的操作如下所示: 在每一个进程开始创建的时候，都会分配一段虚拟地址，然后通过虚拟地址和物理地址的映射来获取真实数据，这样进程就不会直接接触到物理地址，甚至不知道自己调用的哪块物理地址的数据。 目前，大多数操作系统都使用了虚拟内存，如 Windows 系统的虚拟内存、Linux 系统的交换空间等等。Windows 的虚拟内存(pagefile.sys)是磁盘空间的一部分。 在 32 位的系统上，虚拟地址空间大小是 2^32bit=4G。在 64 位系统上，最大虚拟地址空间大小是多少? 是不是 2^64bit=10241014TB=1024PB=16EB?实际上没有用到 64 位，因为用不到这么大的空间，而且会造成很大的系统开销。Linux 一般用低 48 位来表示虚拟地址空间，也就是 2^48bit=256T。 cat /proc/cpuinfo address sizes : 40 bits physical, 48 bits virtual 实际的物理内存可能远远小于虚拟内存的大小。 总结：引入虚拟内存，可以提供更大的地址空间，并且地址空间是连续的，使得程序编写、链接更加简单。并且可以对物理内存进行隔离，不同的进程操作互不影响。还可以通过把同一块物理内存映射到不同的虚拟地址空间实现内存共享。 4.4.2 用户空间和内核空间 为了避免用户进程直接操作内核，保证内核安全，操作系统将虚拟内存划分为两部分，一部分是内核空间(Kernel-space)/ˈkɜːnl /，一部分是用户空间(User-space)。 内核是操作系统的核心，独立于普通的应用程序，可以访问受保护的内存空间，也有访问底层硬件设备的权限。 内核空间中存放的是内核代码和数据，而进程的用户空间中存放的是用户程序的代码和数据。不管是内核空间还是用户空间，它们都处于虚拟空间中，都是对物理地址的映射。 在 Linux 系统中, 内核进程和用户进程所占的虚拟内存比例是 1:3。 当进程运行在内核空间时就处于内核态，而进程运行在用户空间时则处于用户态。 进程在内核空间以执行任意命令，调用系统的一切资源;在用户空间只能执行简单的运算，不能直接调用系统资源，必须通过系统接口(又称 system call)，才能向内核发出指令。 top 命令: us 代表 CPU 消耗在 User space 的时间百分比; sy 代表 CPU 消耗在 Kernel space 的时间百分比。 4.4.3 进程切换(上下文切换) 多任务操作系统是怎么实现运行远大于 CPU 数量的任务个数的? 当然，这些任务实际上并不是真的在同时运行，而是因为系统通过时间片分片算法，在很短的时间内，将 CPU 轮流分配给它们，造成多任务同时运行的错觉。 为了控制进程的执行，内核必须有能力挂起正在 CPU 上运行的进程，并恢复以前挂起的某个进程的执行。这种行为被称为进程切换。 什么叫上下文? 在每个任务运行前，CPU 都需要知道任务从哪里加载、又从哪里开始运行，也就是说，需要系统事先帮它设置好 CPU 寄存器和程序计数器(ProgramCounter)，这个叫做 CPU 的上下文。 而这些保存下来的上下文，会存储在系统内核中，并在任务重新调度执行时再次加载进来。这样就能保证任务原来的状态不受影响，让任务看起来还是连续运行。 在切换上下文的时候，需要完成一系列的工作，这是一个很消耗资源的操作。 4.4.4 进程的阻塞 正在运行的进程由于提出系统服务请求(如 I/O 操作)，但因为某种原因未得到操作系统的立即响应，该进程只能把自己变成阻塞状态，等待相应的事件出现后才被唤醒。 进程在阻塞状态不占用 CPU 资源。 4.4.5 文件描述符 FD Linux 系统将所有设备都当作文件来处理，而 Linux 用文件描述符来标识每个文件对象。 文件描述符(File Descriptor)是内核为了高效管理已被打开的文件所创建的索引，用于指向被打开的文件，所有执行 I/O 操作的系统调用都通过文件描述符;文件描述符是一个简单的非负整数，用以表明每个被进程打开的文件。 Linux 系统里面有三个标准文件描述符。 0:标准输入(键盘); 1:标准输出(显示器); 2:标准错误输出(显示器)。 4.4.6 传统 I/O 数据拷贝 以读操作为例: 当应用程序执行 read 系统调用读取文件描述符(FD)的时候，如果这块数据已经存在于用户进程的页内存中，就直接从内存中读取数据。如果数据不存在，则先将数据从磁盘加载数据到内核缓冲区中，再从内核缓冲区拷贝到用户进程的页内存中。(两次拷贝，两次 user 和 kernel 的上下文切换)。 I/O 的阻塞到底阻塞在哪里? 4.4.7 Blocking I/O 当使用 read 或 write 对某个文件描述符进行过读写时，如果当前 FD 不可读，系统就不会对其他的操作做出响应。从设备复制数据到内核缓冲区是阻塞的，从内核缓冲区拷贝到用户空间，也是阻塞的，直到 copy complete，内核返回结果，用户进程才解除 block 的状态。 为了解决阻塞的问题，我们有几个思路。 1、在服务端创建多个线程或者使用线程池，但是在高并发的情况下需要的线程会很多，系统无法承受，而且创建和释放线程都需要消耗资源。 2、由请求方定期轮询，在数据准备完毕后再从内核缓存缓冲区复制数据到用户空间 (非阻塞式 I/O)，这种方式会存在一定的延迟。 能不能用一个线程处理多个客户端请求? 4.4.8 I/O 多路复用(I/O Multiplexing) I/O 指的是网络 I/O。 多路指的是多个 TCP 连接(Socket 或 Channel)。 复用指的是复用一个或多个线程。它的基本原理就是不再由应用程序自己监视连接，而是由内核替应用程序监视文件描述符。 客户端在操作的时候，会产生具有不同事件类型的 socket。在服务端，I/O 多路复用程序(I/O Multiplexing Module)会把消息放入队列中，然后通过文件事件分派器(File event Dispatcher)，转发到不同的事件处理器中。 多路复用有很多的实现，以 select 为例，当用户进程调用了多路复用器，进程会被阻塞。内核会监视多路复用器负责的所有 socket，当任何一个 socket 的数据准备好了，多路复用器就会返回。这时候用户进程再调用 read 操作，把数据从内核缓冲区拷贝到用户空间。 所以，I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符，而这些文件描述符(套接字描述符)其中的任意一个进入读就绪(readable)状态，select() 函数就可以返回。 Redis 的多路复用， 提供了 select, epoll, evport, kqueue 几种选择，在编译的时 候来选择一种。 evport 是 Solaris 系统内核提供支持的; epoll 是 LINUX 系统内核提供支持的; kqueue 是 Mac 系统提供支持的; select 是 POSIX 提供的，一般的操作系统都有支撑(保底方案); 源码 ae_epoll.c、ae_select.c、ae_kqueue.c、ae_evport.c 5、内存回收 Reids 所有的数据都是存储在内存中的，在某些情况下需要对占用的内存空间进行回 收。内存回收主要分为两类，一类是 key 过期，一类是内存使用达到上限(max_memory) 触发内存淘汰。 5.1 过期策略 要实现 key 过期，我们有几种思路。 5.1.1 定时过期(主动淘汰) 每个设置过期时间的 key 都需要创建一个定时器，到过期时间就会立即清除。该策略可以立即清除过期的数据，对内存很友好;但是会占用大量的 CPU 资源去处理过期的 数据，从而影响缓存的响应时间和吞吐量。 5.1.2 惰性过期(被动淘汰) 只有当访问一个 key 时，才会判断该 key 是否已过期，过期则清除。该策略可以最大化地节省 CPU 资源，却对内存非常不友好。极端情况可能出现大量的过期 key 没有再次被访问，从而不会被清除，占用大量内存。 例如 String，在 getCommand 里面会调用 expireIfNeeded server.c expireIfNeeded(redisDb db, robj key) 第二种情况，每次写入 key 时，发现内存不够，调用 activeExpireCycle 释放一部分内存。 expire.c activeExpireCycle(int type) 5.1.3 定期过期 源码:server.h typedef struct redisDb { dict dict; / 所有的键值对 /dict expires; / 设置了过期时间的键值对 /dict blocking_keys; dict ready_keys; dict watched_keys; int id;long long avg_ttl;list defrag_later; } redisDb; 每隔一定的时间，会扫描一定数量的数据库的 expires 字典中一定数量的 key，并清除其中已过期的 key。该策略是前两者的一个折中方案。通过调整定时扫描的时间间隔和每次扫描的限定耗时，可以在不同情况下使得 CPU 和内存资源达到最优的平衡效果。 Redis 中同时使用了惰性过期和定期过期两种过期策略。 5.2 淘汰策略 Redis 的内存淘汰策略，是指当内存使用达到最大内存极限时，需要使用淘汰算法来决定清理掉哪些数据，以保证新数据的存入。 5.2.1 最大内存设置 redis.conf 参数配置: maxmemory <bytes> 如果不设置 maxmemory 或者设置为 0，64 位系统不限制内存，32 位系统最多使用 3GB 内存。 动态修改: redis> config set maxmemory 2GB 到达最大内存以后怎么办? 5.2.2 淘汰策略 https://redis.io/topics/lru-cache redis.conf maxmemory-policy noeviction 先从算法来看: LRU，Least Recently Used:最近最少使用。判断最近被使用的时间，目前最远的数据优先被淘汰。 LFU，Least Frequently Used，最不常用，4.0 版本新增。 random，随机删除。 如果没有符合前提条件的 key 被淘汰，那么 volatile-lru、volatile-random、 volatile-ttl 相当于 noeviction(不做内存回收)。 动态修改淘汰策略: redis> config set maxmemory-policy volatile-lru 建议使用 volatile-lru，在保证正常服务的情况下，优先删除最近最少使用的 key。 5.2.3 LRU 淘汰原理 问题：如果基于传统 LRU 算法实现 Redis LRU 会有什么问题? 需要额外的数据结构存储，消耗内存。 Redis LRU 对传统的 LRU 算法进行了改良，通过随机采样来调整算法的精度。如果淘汰策略是 LRU，则根据配置的采样值 maxmemory_samples(默认是 5 个), 随机从数据库中选择 m 个 key, 淘汰其中热度最低的 key 对应的缓存数据。所以采样参数m配置的数值越大, 就越能精确的查找到待淘汰的缓存数据,但是也消耗更多的CPU计算,执行效率降低。 问题：如何找出热度最低的数据? Redis 中所有对象结构都有一个 lru 字段, 且使用了 unsigned 的低 24 位，这个字段用来记录对象的热度。对象被创建时会记录 lru 值。在被访问的时候也会更新 lru 的值。 但是不是获取系统当前的时间戳，而是设置为全局变量 server.lruclock 的值。 源码：server.h typedef struct redisObject {unsigned type:4;unsigned encoding:4;unsigned lru:LRU_BITS;int refcount;void ptr; } robj; server.lruclock 的值怎么来的? Redis 中有个定时处理的函数 serverCron，默认每 100 毫秒调用函数 updateCachedTime 更新一次全局变量的 server.lruclock 的值，它记录的是当前 unix 时间戳。 源码:server.c void updateCachedTime(void) { time_t unixtime = time(NULL); atomicSet(server.unixtime,unixtime); server.mstime = mstime();​struct tm tm; localtime_r(&server.unixtime,&tm);server.daylight_active = tm.tm_isdst; } 问题:为什么不获取精确的时间而是放在全局变量中?不会有延迟的问题吗? 这样函数 lookupKey 中更新数据的 lru 热度值时,就不用每次调用系统函数 time，可以提高执行效率。 OK，当对象里面已经有了 LRU 字段的值，就可以评估对象的热度了。 函数 estimateObjectIdleTime 评估指定对象的 lru 热度，思想就是对象的 lru 值和全局的 server.lruclock 的差值越大(越久没有得到更新)，该对象热度越低。 源码 evict.c / Given an object returns the min number of milliseconds the object was never requested, using an approximated LRU algorithm. /unsigned long long estimateObjectIdleTime(robj o) {unsigned long long lruclock = LRU_CLOCK(); if (lruclock >= o->lru) {return (lruclock - o->lru) LRU_CLOCK_RESOLUTION; } else {return (lruclock + (LRU_CLOCK_MAX - o->lru)) LRU_CLOCK_RESOLUTION;} } server.lruclock 只有 24 位，按秒为单位来表示才能存储 194 天。当超过 24bit 能表 示的最大时间的时候，它会从头开始计算。 server.h define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) / Max value of obj->lru / 在这种情况下，可能会出现对象的 lru 大于 server.lruclock 的情况，如果这种情况 出现那么就两个相加而不是相减来求最久的 key。 为什么不用常规的哈希表+双向链表的方式实现?需要额外的数据结构，消耗资源。而 Redis LRU 算法在 sample 为 10 的情况下，已经能接近传统 LRU 算法了。 问题:除了消耗资源之外，传统 LRU 还有什么问题? 如图，假设 A 在 10 秒内被访问了 5 次，而 B 在 10 秒内被访问了 3 次。因为 B 最后一次被访问的时间比 A 要晚，在同等的情况下，A 反而先被回收。 问题:要实现基于访问频率的淘汰机制，怎么做? 5.2.4 LFU server.h typedef struct redisObject {unsigned type:4;unsigned encoding:4;unsigned lru:LRU_BITS;int refcount;void ptr; } robj; 当这 24 bits 用作 LFU 时，其被分为两部分: 高 16 位用来记录访问时间(单位为分钟，ldt，last decrement time) 低 8 位用来记录访问频率，简称 counter(logc，logistic counter) counter 是用基于概率的对数计数器实现的，8 位可以表示百万次的访问频率。 对象被读写的时候，lfu 的值会被更新。 db.c——lookupKey void updateLFU(robj val) {unsigned long counter = LFUDecrAndReturn(val); counter = LFULogIncr(counter);val->lru = (LFUGetTimeInMinutes()<<8) | counter;} 增长的速率由，lfu-log-factor 越大，counter 增长的越慢 redis.conf 配置文件。 lfu-log-factor 10 如果计数器只会递增不会递减，也不能体现对象的热度。没有被访问的时候，计数器怎么递减呢? 减少的值由衰减因子 lfu-decay-time(分钟)来控制，如果值是 1 的话，N 分钟没有访问就要减少 N。 redis.conf 配置文件 lfu-decay-time 1 6、持久化机制 https://redis.io/topics/persistence Redis 速度快，很大一部分原因是因为它所有的数据都存储在内存中。如果断电或者宕机，都会导致内存中的数据丢失。为了实现重启后数据不丢失，Redis 提供了两种持久化的方案，一种是 RDB 快照(Redis DataBase)，一种是 AOF(Append Only File)。 6.1 RDB RDB 是 Redis 默认的持久化方案。当满足一定条件的时候，会把当前内存中的数据写入磁盘，生成一个快照文件 dump.rdb。Redis 重启会通过加载 dump.rdb 文件恢复数据。 什么时候写入 rdb 文件? 6.1.1 RDB 触发 1、自动触发 a)配置规则触发。 redis.conf， SNAPSHOTTING，其中定义了触发把数据保存到磁盘的触发频率。 如果不需要 RDB 方案，注释 save 或者配置成空字符串""。 save 900 1 900 秒内至少有一个 key 被修改(包括添加) save 300 10 400 秒内至少有 10 个 key 被修改save 60 10000 60 秒内至少有 10000 个 key 被修改 注意上面的配置是不冲突的，只要满足任意一个都会触发。 RDB 文件位置和目录: 文件路径，dir ./ 文件名称dbfilename dump.rdb 是否是LZF压缩rdb文件 rdbcompression yes 开启数据校验 rdbchecksum yes 问题：为什么停止 Redis 服务的时候没有 save，重启数据还在? RDB 还有两种触发方式: b)shutdown 触发，保证服务器正常关闭。 c)flushall，RDB 文件是空的，没什么意义(删掉 dump.rdb 演示一下)。 2、手动触发 如果我们需要重启服务或者迁移数据，这个时候就需要手动触 RDB 快照保存。Redis 提供了两条命令: a)save save 在生成快照的时候会阻塞当前 Redis 服务器， Redis 不能处理其他命令。如果内存中的数据比较多，会造成 Redis 长时间的阻塞。生产环境不建议使用这个命令。 为了解决这个问题，Redis 提供了第二种方式。 执行 bgsave 时，Redis 会在后台异步进行快照操作，快照同时还可以响应客户端请求。 具体操作是 Redis 进程执行 fork 操作创建子进程(copy-on-write)，RDB 持久化过程由子进程负责，完成后自动结束。它不会记录 fork 之后后续的命令。阻塞只发生在 fork 阶段，一般时间很短。 用 lastsave 命令可以查看最近一次成功生成快照的时间。 6.1.2 RDB 数据的恢复(演示) 1、shutdown 持久化添加键值 添加键值 redis> set k1 1 redis> set k2 2 redis> set k3 3 redis> set k4 4 redis> set k5 5 停服务器，触发 save redis> shutdown 备份 dump.rdb 文件 cp dump.rdb dump.rdb.bak 启动服务器 /usr/local/soft/redis-5.0.5/src/redis-server /usr/local/soft/redis-5.0.5/redis.conf 啥都没有: redis> keys 3、通过备份文件恢复数据停服务器 redis> shutdown 重命名备份文件 mv dump.rdb.bak dump.rdb 启动服务器 /usr/local/soft/redis-5.0.5/src/redis-server /usr/local/soft/redis-5.0.5/redis.conf 查看数据 redis> keys 6.1.3 RDB 文件的优势和劣势 一、优势 1.RDB 是一个非常紧凑(compact)的文件，它保存了 redis 在某个时间点上的数据集。这种文件非常适合用于进行备份和灾难恢复。 2.生成 RDB 文件的时候，redis 主进程会 fork()一个子进程来处理所有保存工作，主进程不需要进行任何磁盘 IO 操作。 3.RDB 在恢复大数据集时的速度比 AOF 的恢复速度要快。 二、劣势 1、RDB 方式数据没办法做到实时持久化/秒级持久化。因为 bgsave 每次运行都要执行 fork 操作创建子进程，频繁执行成本过高。 2、在一定间隔时间做一次备份，所以如果 redis 意外 down 掉的话，就会丢失最后一次快照之后的所有修改(数据有丢失)。 如果数据相对来说比较重要，希望将损失降到最小，则可以使用 AOF 方式进行持久化。 6.2 AOF Append Only File AOF:Redis 默认不开启。AOF 采用日志的形式来记录每个写操作，并追加到文件中。开启后，执行更改 Redis 数据的命令时，就会把命令写入到 AOF 文件中。 Redis 重启时会根据日志文件的内容把写指令从前到后执行一次以完成数据的恢复工作。 6.2.1 AOF 配置 配置文件 redis.conf 开关appendonly no 文件名appendfilename "appendonly.aof" AOF 文件的内容(vim 查看): 问题：数据都是实时持久化到磁盘吗? 由于操作系统的缓存机制，AOF 数据并没有真正地写入硬盘，而是进入了系统的硬盘缓存。什么时候把缓冲区的内容写入到 AOF 文件? 问题:文件越来越大，怎么办? 由于 AOF 持久化是 Redis 不断将写命令记录到 AOF 文件中，随着 Redis 不断的进行，AOF 的文件会越来越大，文件越大，占用服务器内存越大以及 AOF 恢复要求时间越长。 例如 set xxx 666，执行 1000 次，结果都是 xxx=666。 为了解决这个问题，Redis 新增了重写机制，当 AOF 文件的大小超过所设定的阈值时，Redis 就会启动 AOF 文件的内容压缩，只保留可以恢复数据的最小指令集。 可以使用命令 bgrewriteaof 来重写。 AOF 文件重写并不是对原文件进行重新整理，而是直接读取服务器现有的键值对，然后用一条命令去代替之前记录这个键值对的多条命令，生成一个新的文件后去替换原来的 AOF 文件。 重写触发机制 auto-aof-rewrite-percentage 100 auto-aof-rewrite-min-size 64mb 问题:重写过程中，AOF 文件被更改了怎么办? 另外有两个与 AOF 相关的参数: 6.2.2 AOF 数据恢复 重启 Redis 之后就会进行 AOF 文件的恢复。 6.2.3 AOF 优势与劣势 优点: 1、AOF 持久化的方法提供了多种的同步频率，即使使用默认的同步频率每秒同步一次，Redis 最多也就丢失 1 秒的数据而已。 缺点: 1、对于具有相同数据的的 Redis，AOF 文件通常会比 RDB 文件体积更大(RDB 存的是数据快照)。 2、虽然 AOF 提供了多种同步的频率，默认情况下，每秒同步一次的频率也具有较高的性能。在高并发的情况下，RDB 比 AOF 具好更好的性能保证。 6.3 两种方案比较 那么对于 AOF 和 RDB 两种持久化方式，我们应该如何选择呢? 如果可以忍受一小段时间内数据的丢失，毫无疑问使用 RDB 是最好的，定时生成 RDB 快照(snapshot)非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快。 否则就使用 AOF 重写。但是一般情况下建议不要单独使用某一种持久化机制，而是应该两种一起用，在这种情况下,当 redis 重启的时候会优先载入 AOF 文件来恢复原始的数据，因为在通常情况下 AOF 文件保存的数据集要比 RDB 文件保存的数据集要完整。 本篇文章为转载内容。原文链接：https://blog.csdn.net/zhoutaochun/article/details/120075092。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...。本文将详细介绍如何解决这些问题。 二、环境变量未正确设置 环境变量未正确设置是导致Dubbo无法正常运行的一个重要原因。比如说，如果你没把JAVA_HOME环境变量设置对，Dubbo就找不到Java的藏身之处（也就是安装路径），这样一来，它就没法正常启动运行啦。 解决这个问题的方法非常简单，只需要在系统环境变量中添加JAVA_HOME即可。例如，在Windows系统中，可以在"我的电脑" -> "属性" -> "高级系统设置" -> "环境变量"中添加。 三、日志配置错误 日志配置错误也是导致Dubbo无法正常运行的一个重要原因。要是你日志的配置文件，比如说logback.xml，搞错了设定，那就等于给日志输出挖了个坑。这样一来，日志就无法顺畅地“说话”了，我们也就没法通过这些日志来摸清系统的运行状况，了解它到底是怎么干活儿的了。 解决这个问题的方法也很简单，只需要检查日志配置文件中的配置是否正确即可。比如，我们可以瞅瞅日志输出的目的地是不是设定对了，还有日志的详细程度级别是否也调得恰到好处，这些小细节都值得我们关注检查一下。 四、代码示例 为了更直观地理解环境配置问题和日志配置错误，下面给出一些代码示例。 首先，来看一下不正确的环境变量设置。假设我们在没有设置JAVA_HOME的情况下尝试启动Dubbo，那么就会出现以下错误： Exception in thread "main" java.lang.UnsatisfiedLinkError: no javassist in java.library.path at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1867) at java.lang.Runtime.loadLibrary0(Runtime.java:870) at java.lang.System.loadLibrary(System.java:1122) at com.alibaba.dubbo.common.logger.LoggerFactory.getLogger(LoggerFactory.java:39) at com.alibaba.dubbo.common.logger.LoggerFactory.getLogger(LoggerFactory.java:51) at com.alibaba.dubbo.config.ApplicationConfig.(ApplicationConfig.java:114) at com.example.demo.DemoApplication.main(DemoApplication.java:12) Caused by: java.lang.ClassNotFoundException: javassist at java.net.URLClassLoader.findClass(URLClassLoader.java:382) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ... 6 more 可以看出，由于JAVA_HOME环境变量未设置，所以无法找到Java的安装路径，从而导致了这个错误。 接下来，来看一下不正确的日志配置。假设我们在日志配置文件中错误地指定了日志输出的目标位置，那么就会出现以下错误： 2022-03-08 15:29:54,742 ERROR [main] org.apache.log4j.ConsoleAppender - Error initializing ConsoleAppender appenders named [STDOUT] org.apache.log4j.AppenderSkeleton$InvalidAppenderException: No such appender 'STDOUT' in category [com.example.demo]. at org.apache.log4j.Category.forcedLog(Category.java:393) at org.apache.log4j.Category.access$100(Category.java:67) at org.apache.log4j.Category$AppenderAttachedObject.append(Category.java:839) at org.apache.log4j.AppenderSkeleton.doAppend(AppenderSkeleton.java:248) at org.apache.log4j.helpers.AppenderAttachableImpl.appendLoopOnAppenders(AppenderAttachableImpl.java:51) at org.apache.log4j.Category.callAppenders(Category.java:206) at org.apache.log4j.Category.debug(Category.java:267) at org.apache.log4j.Category.info(Category.java:294) at org.apache.log4j.Logger.info(Logger.java:465) at com.example.demo.DemoApplication.main(DemoApplication.java:16) 可以看出，由于日志配置文件中的配置错误，所以无法将日志输出到指定的位置，从而导致了这个错误。 五、总结 通过以上分析，我们可以看出，环境配置问题和日志配置错误都是非常严重的问题，如果不及时处理，就会导致Dubbo无法正常运行，从而影响我们的工作。所以呢，咱们得好好学习、掌握这些知识点，这样一来，在实际工作中碰到问题时，就能更有效率地避开陷阱，解决麻烦了。同时，我们也应该养成良好的编程习惯，比如定期检查环境变量和日志配置文件，确保它们的正确性。

...ller）设计模式的Java Web应用程序框架，它主要用于简化构建企业级Java Web应用的工作。在本文中，Struts被用来实现文件上传和下载功能，通过定义Action类、配置struts.xml文件以及使用拦截器等机制，实现了对HTTP请求的接收、处理及响应。 MIME类型（Multipurpose Internet Mail Extensions） , MIME类型是一种标准，用于指定数据内容的格式类型，如文本、图像、视频或应用程序特定的数据。在Web开发中，特别是文件上传和下载场景，服务器端和客户端需要根据MIME类型来正确解析和处理不同类型的文件。例如，在Struts框架中，通过配置MIME类型可以指示浏览器如何打开或保存从服务器下载的文件。 拦截器（Interceptor） , 在Struts 2框架中，拦截器是一个可插拔的对象，它可以参与到Action执行的整个生命周期中，并在特定阶段进行预处理或后处理操作。文章中的LoginInterceptor就是一个自定义拦截器，它负责检查用户是否已经登录，只有当用户已登录时才允许继续执行后续的操作（如文件上传或下载）。通过这种方式，拦截器增强了系统的安全性，确保了只有经过验证的用户才能访问受限资源。

...ket往后台发送日志数据，在这里我们是要做基于SparkStreaming做实时在线统计。那么数据就需要放进消息系统（Kafka）中，我们的Spark Streaming应用程序就会去Kafka中Pull数据过来进行计算和消费，并把计算后的数据放入到持久化系统中（MySQL） 广告点击系统实时分析的意义：因为可以在线实时的看见广告的投放效果，就为广告的更大规模的投入和调整打下了坚实的基础，从而为公司带来最大化的经济回报。 核心需求： 1、实时黑名单动态过滤出有效的用户广告点击行为：因为黑名单用户可能随时出现，所以需要动态更新； 2、在线计算广告点击流量； 3、Top3热门广告； 4、每个广告流量趋势； 5、广告点击用户的区域分布分析 6、最近一分钟的广告点击量； 7、整个广告点击Spark Streaming处理程序724小时运行； 数据格式： 时间、用户、广告、城市等 技术细节： 在线计算用户点击的次数分析，屏蔽IP等； 使用updateStateByKey或者mapWithState进行不同地区广告点击排名的计算； Spark Streaming+Spark SQL+Spark Core等综合分析数据； 使用Window类型的操作； 高可用和性能调优等等； 流量趋势，一般会结合DB等； Spark Core / /package com.tom.spark.SparkApps.sparkstreaming;import java.util.Date;import java.util.HashMap;import java.util.Map;import java.util.Properties;import java.util.Random;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;/ 数据生成代码，Kafka Producer产生数据/public class MockAdClickedStat {/ @param args/public static void main(String[] args) {final Random random = new Random();final String[] provinces = new String[]{"Guangdong", "Zhejiang", "Jiangsu", "Fujian"};final Map<String, String[]> cities = new HashMap<String, String[]>();cities.put("Guangdong", new String[]{"Guangzhou", "Shenzhen", "Dongguan"});cities.put("Zhejiang", new String[]{"Hangzhou", "Wenzhou", "Ningbo"});cities.put("Jiangsu", new String[]{"Nanjing", "Suzhou", "Wuxi"});cities.put("Fujian", new String[]{"Fuzhou", "Xiamen", "Sanming"});final String[] ips = new String[] {"192.168.112.240","192.168.112.239","192.168.112.245","192.168.112.246","192.168.112.247","192.168.112.248","192.168.112.249","192.168.112.250","192.168.112.251","192.168.112.252","192.168.112.253","192.168.112.254",};/ Kafka相关的基本配置信息/Properties kafkaConf = new Properties();kafkaConf.put("serializer.class", "kafka.serializer.StringEncoder");kafkaConf.put("metadeta.broker.list", "Master:9092,Worker1:9092,Worker2:9092");ProducerConfig producerConfig = new ProducerConfig(kafkaConf);final Producer<Integer, String> producer = new Producer<Integer, String>(producerConfig);new Thread(new Runnable() {public void run() {while(true) {//在线处理广告点击流的基本数据格式：timestamp、ip、userID、adID、province、cityLong timestamp = new Date().getTime();String ip = ips[random.nextInt(12)]; //可以采用网络上免费提供的ip库int userID = random.nextInt(10000);int adID = random.nextInt(100);String province = provinces[random.nextInt(4)];String city = cities.get(province)[random.nextInt(3)];String clickedAd = timestamp + "\t" + ip + "\t" + userID + "\t" + adID + "\t" + province + "\t" + city;producer.send(new KeyedMessage<Integer, String>("AdClicked", clickedAd));try {Thread.sleep(50);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }).start();} } package com.tom.spark.SparkApps.sparkstreaming;import java.sql.Connection;import java.sql.DriverManager;import java.sql.PreparedStatement;import java.sql.ResultSet;import java.sql.SQLException;import java.util.ArrayList;import java.util.Arrays;import java.util.HashMap;import java.util.HashSet;import java.util.Iterator;import java.util.List;import java.util.Map;import java.util.Set;import java.util.concurrent.LinkedBlockingQueue;import kafka.serializer.StringDecoder;import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.Function;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import org.apache.spark.api.java.function.VoidFunction;import org.apache.spark.sql.DataFrame;import org.apache.spark.sql.Row;import org.apache.spark.sql.RowFactory;import org.apache.spark.sql.hive.HiveContext;import org.apache.spark.sql.types.DataTypes;import org.apache.spark.sql.types.StructType;import org.apache.spark.streaming.Durations;import org.apache.spark.streaming.api.java.JavaDStream;import org.apache.spark.streaming.api.java.JavaPairDStream;import org.apache.spark.streaming.api.java.JavaPairInputDStream;import org.apache.spark.streaming.api.java.JavaStreamingContext;import org.apache.spark.streaming.api.java.JavaStreamingContextFactory;import org.apache.spark.streaming.kafka.KafkaUtils;import com.google.common.base.Optional;import scala.Tuple2;/ 数据处理，Kafka消费者/public class AdClickedStreamingStats {/ @param args/public static void main(String[] args) {// TODO Auto-generated method stub//好处：1、checkpoint 2、工厂final SparkConf conf = new SparkConf().setAppName("SparkStreamingOnKafkaDirect").setMaster("hdfs://Master:7077/");final String checkpointDirectory = "hdfs://Master:9000/library/SparkStreaming/CheckPoint_Data";JavaStreamingContextFactory factory = new JavaStreamingContextFactory() {public JavaStreamingContext create() {// TODO Auto-generated method stubreturn createContext(checkpointDirectory, conf);} };/ 可以从失败中恢复Driver，不过还需要指定Driver这个进程运行在Cluster，并且在提交应用程序的时候制定--supervise;/JavaStreamingContext javassc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory);/ 第三步：创建Spark Streaming输入数据来源input Stream: 1、数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等 2、在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口的数据 (当然该端口服务首先必须存在），并且在后续会根据业务需要不断有数据产生（当然对于Spark Streaming 应用程序的运行而言，有无数据其处理流程都是一样的） 3、如果经常在每间隔5秒钟没有数据的话不断启动空的Job其实会造成调度资源的浪费，因为并没有数据需要发生计算；所以 实际的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；///创建Kafka元数据来让Spark Streaming这个Kafka Consumer利用Map<String, String> kafkaParameters = new HashMap<String, String>();kafkaParameters.put("metadata.broker.list", "Master:9092,Worker1:9092,Worker2:9092");Set<String> topics = new HashSet<String>();topics.add("SparkStreamingDirected");JavaPairInputDStream<String, String> adClickedStreaming = KafkaUtils.createDirectStream(javassc, String.class, String.class, StringDecoder.class, StringDecoder.class,kafkaParameters, topics);/因为要对黑名单进行过滤，而数据是在RDD中的，所以必然使用transform这个函数； 但是在这里我们必须使用transformToPair，原因是读取进来的Kafka的数据是Pair<String,String>类型, 另一个原因是过滤后的数据要进行进一步处理，所以必须是读进的Kafka数据的原始类型 在此再次说明，每个Batch Duration中实际上讲输入的数据就是被一个且仅被一个RDD封装的，你可以有多个 InputDStream，但其实在产生job的时候，这些不同的InputDStream在Batch Duration中就相当于Spark基于HDFS 数据操作的不同文件来源而已罢了。/JavaPairDStream<String, String> filteredadClickedStreaming = adClickedStreaming.transformToPair(new Function<JavaPairRDD<String,String>, JavaPairRDD<String,String>>() {public JavaPairRDD<String, String> call(JavaPairRDD<String, String> rdd) throws Exception {/ 在线黑名单过滤思路步骤： 1、从数据库中获取黑名单转换成RDD，即新的RDD实例封装黑名单数据； 2、然后把代表黑名单的RDD的实例和Batch Duration产生的RDD进行Join操作， 准确的说是进行leftOuterJoin操作，也就是说使用Batch Duration产生的RDD和代表黑名单的RDD实例进行 leftOuterJoin操作，如果两者都有内容的话，就会是true，否则的话就是false 我们要留下的是leftOuterJoin结果为false； /final List<String> blackListNames = new ArrayList<String>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doQuery("SELECT FROM blacklisttable", null, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {while(result.next()){blackListNames.add(result.getString(1));} }});List<Tuple2<String, Boolean>> blackListTuple = new ArrayList<Tuple2<String,Boolean>>();for(String name : blackListNames) {blackListTuple.add(new Tuple2<String, Boolean>(name, true));}List<Tuple2<String, Boolean>> blacklistFromListDB = blackListTuple; //数据来自于查询的黑名单表并且映射成为<String, Boolean>JavaSparkContext jsc = new JavaSparkContext(rdd.context());/ 黑名单的表中只有userID，但是如果要进行join操作的话就必须是Key-Value，所以在这里我们需要 基于数据表中的数据产生Key-Value类型的数据集合/JavaPairRDD<String, Boolean> blackListRDD = jsc.parallelizePairs(blacklistFromListDB);/ 进行操作的时候肯定是基于userID进行join，所以必须把传入的rdd进行mapToPair操作转化成为符合格式的RDD/JavaPairRDD<String, Tuple2<String, String>> rdd2Pair = rdd.mapToPair(new PairFunction<Tuple2<String,String>, String, Tuple2<String, String>>() {public Tuple2<String, Tuple2<String, String>> call(Tuple2<String, String> t) throws Exception {// TODO Auto-generated method stubString userID = t._2.split("\t")[2];return new Tuple2<String, Tuple2<String,String>>(userID, t);} });JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> joined = rdd2Pair.leftOuterJoin(blackListRDD);JavaPairRDD<String, String> result = joined.filter(new Function<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, Boolean>() {public Boolean call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> tuple)throws Exception {// TODO Auto-generated method stubOptional<Boolean> optional = tuple._2._2;if(optional.isPresent() && optional.get()){return false;} else {return true;} }}).mapToPair(new PairFunction<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, String, String>() {public Tuple2<String, String> call(Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> t)throws Exception {// TODO Auto-generated method stubreturn t._2._1;} });return result;} });//广告点击的基本数据格式：timestamp、ip、userID、adID、province、cityJavaPairDStream<String, Long> pairs = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t) throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} });/ 第4.3步：在单词实例计数为1基础上，统计每个单词在文件中出现的总次数/JavaPairDStream<String, Long> adClickedUsers= pairs.reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long i1, Long i2) throws Exception{return i1 + i2;} });/判断有效的点击，复杂化的采用机器学习训练模型进行在线过滤 简单的根据ip判断1天不超过100次；也可以通过一个batch duration的点击次数判断是否非法广告点击，通过一个batch来判断是不完整的，还需要一天的数据也可以每一个小时来判断。/JavaPairDStream<String, Long> filterClickedBatch = adClickedUsers.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {if (1 < v1._2){//更新一些黑名单的数据库表return false;} else { return true;} }});//filterClickedBatch.print();//写入数据库filterClickedBatch.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：userID,adID,clickedCount,time//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<UserAdClicked> userAdClickedList = new ArrayList<UserAdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");UserAdClicked userClicked = new UserAdClicked();userClicked.setTimestamp(splited[0]);userClicked.setIp(splited[1]);userClicked.setUserID(splited[2]);userClicked.setAdID(splited[3]);userClicked.setProvince(splited[4]);userClicked.setCity(splited[5]);userAdClickedList.add(userClicked);}final List<UserAdClicked> inserting = new ArrayList<UserAdClicked>();final List<UserAdClicked> updating = new ArrayList<UserAdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final UserAdClicked clicked : userAdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclicked WHERE"+ " timestamp =? AND userID = ? AND adID = ?",new Object[]{clicked.getTimestamp(), clicked.getUserID(),clicked.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(UserAdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getIp(),insertRecord.getUserID(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclicked VALUES(?, ?, ?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(UserAdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getTimestamp(),updateRecord.getIp(),updateRecord.getUserID(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity(),updateRecord.getClickedCount() + 1});}jdbcWrapper.doBatch("UPDATE adclicked SET clickedCount = ? WHERE"+ " timestamp =? AND ip = ? AND userID = ? AND adID = ? "+ "AND province = ? AND city = ?", updateParametersList);} });return null;} });//再次过滤，从数据库中读取数据过滤黑名单JavaPairDStream<String, Long> blackListBasedOnHistory = filterClickedBatch.filter(new Function<Tuple2<String,Long>, Boolean>() {public Boolean call(Tuple2<String, Long> v1) throws Exception {//广告点击的基本数据格式：timestamp,ip,userID,adID,province,cityString[] splited = v1._1.split("\t"); //提取key值String date =splited[0];String userID =splited[2];String adID =splited[3];//查询一下数据库同一个用户同一个广告id点击量超过50次列入黑名单//接下来 根据date、userID、adID条件去查询用户点击广告的数据表，获得总的点击次数//这个时候基于点击次数判断是否属于黑名单点击int clickedCountTotalToday = 81 ;if (clickedCountTotalToday > 50) {return true;}else {return false ;} }});//map操作，找出用户的idJavaDStream<String> blackListuserIDBasedInBatchOnhistroy =blackListBasedOnHistory.map(new Function<Tuple2<String,Long>, String>() {public String call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubreturn v1._1.split("\t")[2];} });//有一个问题，数据可能重复，在一个partition里面重复，这个好办；//但多个partition不能保证一个用户重复，需要对黑名单的整个rdd进行去重操作。//rdd去重了，partition也就去重了，一石二鸟，一箭双雕// 找出了黑名单，下一步就写入黑名单数据库表中JavaDStream<String> blackListUniqueuserBasedInBatchOnhistroy = blackListuserIDBasedInBatchOnhistroy.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception {// TODO Auto-generated method stubreturn rdd.distinct();} });// 下一步写入到数据表中blackListUniqueuserBasedInBatchOnhistroy.foreachRDD(new Function<JavaRDD<String>, Void>() {public Void call(JavaRDD<String> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<String>>() {public void call(Iterator<String> t) throws Exception {// TODO Auto-generated method stub//插入的用户信息可以只包含：useID//此时直接插入黑名单数据表即可。//写入数据库List<Object[]> blackList = new ArrayList<Object[]>();while(t.hasNext()) {blackList.add(new Object[]{t.next()});}JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();jdbcWrapper.doBatch("INSERT INTO blacklisttable values (?)", blackList);} });return null;} });/广告点击累计动态更新,每个updateStateByKey都会在Batch Duration的时间间隔的基础上进行广告点击次数的更新， 更新之后我们一般都会持久化到外部存储设备上，在这里我们存储到MySQL数据库中/JavaPairDStream<String, Long> updateStateByKeyDSteam = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String[] splited=t._2.split("\t");String timestamp = splited[0]; //YYYY-MM-DDString ip = splited[1];String userID = splited[2];String adID = splited[3];String province = splited[4];String city = splited[5]; String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"+province +"_"+city;return new Tuple2<String, Long>(clickedRecord, 1L);} }).updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {public Optional<Long> call(List<Long> v1, Optional<Long> v2)throws Exception {// v1:当前的Key在当前的Batch Duration中出现的次数的集合，例如{1，1，1，。。。，1}// v2:当前的Key在以前的Batch Duration中积累下来的结果；Long clickedTotalHistory = 0L; if(v2.isPresent()){clickedTotalHistory = v2.get();}for(Long one : v1) {clickedTotalHistory += one;}return Optional.of(clickedTotalHistory);} });updateStateByKeyDSteam.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql//例如一次插入 1000条 records，使用insertBatch 或 updateBatch//插入的用户数据信息：timestamp、adID、province、city//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作List<AdClicked> AdClickedList = new ArrayList<AdClicked>();while(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("\t");AdClicked adClicked = new AdClicked();adClicked.setTimestamp(splited[0]);adClicked.setAdID(splited[1]);adClicked.setProvince(splited[2]);adClicked.setCity(splited[3]);adClicked.setClickedCount(record._2);AdClickedList.add(adClicked);}final List<AdClicked> inserting = new ArrayList<AdClicked>();final List<AdClicked> updating = new ArrayList<AdClicked>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdClicked clicked : AdClickedList) {jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedcount WHERE"+ " timestamp = ? AND adID = ? AND province = ? AND city = ?",new Object[]{clicked.getTimestamp(), clicked.getAdID(),clicked.getProvince(), clicked.getCity()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);clicked.setClickedCount(count);updating.add(clicked);} else {inserting.add(clicked);clicked.setClickedCount(1L);} }});}//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdClicked insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince(),insertRecord.getCity(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedcount VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdClicked updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.getTimestamp(),updateRecord.getAdID(),updateRecord.getProvince(),updateRecord.getCity()});}jdbcWrapper.doBatch("UPDATE adclickedcount SET clickedCount = ? WHERE"+ " timestamp =? AND adID = ? AND province = ? AND city = ?", updateParametersList);} });return null;} });/ 对广告点击进行TopN计算，计算出每天每个省份Top5排名的广告 因为我们直接对RDD进行操作，所以使用了transfomr算子；/updateStateByKeyDSteam.transform(new Function<JavaPairRDD<String,Long>, JavaRDD<Row>>() {public JavaRDD<Row> call(JavaPairRDD<String, Long> rdd) throws Exception {JavaRDD<Row> rowRDD = rdd.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, Long> t)throws Exception {// TODO Auto-generated method stubString[] splited=t._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];String clickedRecord = timestamp + "_" + adID + "_" + province;return new Tuple2<String, Long>(clickedRecord, t._2);} }).reduceByKey(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }).map(new Function<Tuple2<String,Long>, Row>() {public Row call(Tuple2<String, Long> v1) throws Exception {// TODO Auto-generated method stubString[] splited=v1._1.split("_");String timestamp = splited[0]; //YYYY-MM-DDString adID = splited[3];String province = splited[4];return RowFactory.create(timestamp, adID, province, v1._2);} });StructType structType = DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("timestamp", DataTypes.StringType, true),DataTypes.createStructField("adID", DataTypes.StringType, true),DataTypes.createStructField("province", DataTypes.StringType, true),DataTypes.createStructField("clickedCount", DataTypes.LongType, true)));HiveContext hiveContext = new HiveContext(rdd.context());DataFrame df = hiveContext.createDataFrame(rowRDD, structType);df.registerTempTable("topNTableSource");DataFrame result = hiveContext.sql("SELECT timestamp, adID, province, clickedCount, FROM"+ " (SELECT timestamp, adID, province,clickedCount, "+ "ROW_NUMBER() OVER(PARTITION BY province ORDER BY clickeCount DESC) rank "+ "FROM topNTableSource) subquery "+ "WHERE rank <= 5");return result.toJavaRDD();} }).foreachRDD(new Function<JavaRDD<Row>, Void>() {public Void call(JavaRDD<Row> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Row>>() {public void call(Iterator<Row> t) throws Exception {// TODO Auto-generated method stubList<AdProvinceTopN> adProvinceTopN = new ArrayList<AdProvinceTopN>();while(t.hasNext()) {Row row = t.next();AdProvinceTopN item = new AdProvinceTopN();item.setTimestamp(row.getString(0));item.setAdID(row.getString(1));item.setProvince(row.getString(2));item.setClickedCount(row.getLong(3));adProvinceTopN.add(item);}// final List<AdProvinceTopN> inserting = new ArrayList<AdProvinceTopN>();// final List<AdProvinceTopN> updating = new ArrayList<AdProvinceTopN>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();Set<String> set = new HashSet<String>();for(AdProvinceTopN item: adProvinceTopN){set.add(item.getTimestamp() + "_" + item.getProvince());}//表的字段timestamp、adID、province、clickedCountArrayList<Object[]> deleteParametersList = new ArrayList<Object[]>();for(String deleteRecord : set) {String[] splited = deleteRecord.split("_");deleteParametersList.add(new Object[]{splited[0],splited[1]});}jdbcWrapper.doBatch("DELETE FROM adprovincetopn WHERE timestamp = ? AND province = ?", deleteParametersList);//表的字段timestamp、ip、userID、adID、province、city、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdProvinceTopN insertRecord : adProvinceTopN) {insertParametersList.add(new Object[] {insertRecord.getClickedCount(),insertRecord.getTimestamp(),insertRecord.getAdID(),insertRecord.getProvince()});}jdbcWrapper.doBatch("INSERT INTO adprovincetopn VALUES (?, ?, ?, ?)", insertParametersList);} });return null;} });/ 计算过去半个小时内广告点击的趋势 广告点击的基本数据格式：timestamp、ip、userID、adID、province、city/filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {public Tuple2<String, Long> call(Tuple2<String, String> t)throws Exception {String splited[] = t._2.split("\t");String adID = splited[3];String time = splited[0]; //Todo:后续需要重构代码实现时间戳和分钟的转换提取。此处需要提取出该广告的点击分钟单位return new Tuple2<String, Long>(time + "_" + adID, 1L);} }).reduceByKeyAndWindow(new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 + v2;} }, new Function2<Long, Long, Long>() {public Long call(Long v1, Long v2) throws Exception {// TODO Auto-generated method stubreturn v1 - v2;} }, Durations.minutes(30), Durations.milliseconds(5)).foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {public Void call(JavaPairRDD<String, Long> rdd) throws Exception {// TODO Auto-generated method stubrdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {public void call(Iterator<Tuple2<String, Long>> partition)throws Exception {List<AdTrendStat> adTrend = new ArrayList<AdTrendStat>();// TODO Auto-generated method stubwhile(partition.hasNext()) {Tuple2<String, Long> record = partition.next();String[] splited = record._1.split("_");String time = splited[0];String adID = splited[1];Long clickedCount = record._2;/ 在插入数据到数据库的时候具体需要哪些字段？time、adID、clickedCount; 而我们通过J2EE技术进行趋势绘图的时候肯定是需要年、月、日、时、分这个维度的，所以我们在这里需要 年月日、小时、分钟这些时间维度；/AdTrendStat adTrendStat = new AdTrendStat();adTrendStat.setAdID(adID);adTrendStat.setClickedCount(clickedCount);adTrendStat.set_date(time); //Todo:获取年月日adTrendStat.set_hour(time); //Todo:获取小时adTrendStat.set_minute(time);//Todo:获取分钟adTrend.add(adTrendStat);}final List<AdTrendStat> inserting = new ArrayList<AdTrendStat>();final List<AdTrendStat> updating = new ArrayList<AdTrendStat>();JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();//表的字段timestamp、ip、userID、adID、province、city、clickedCountfor(final AdTrendStat trend : adTrend) {final AdTrendCountHistory adTrendhistory = new AdTrendCountHistory();jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedtrend WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?",new Object[]{trend.get_date(), trend.get_hour(), trend.get_minute(),trend.getAdID()}, new ExecuteCallBack() {public void resultCallBack(ResultSet result) throws Exception {// TODO Auto-generated method stubif(result.next()) {long count = result.getLong(1);adTrendhistory.setClickedCountHistoryLong(count);updating.add(trend);} else { inserting.add(trend);} }});}//表的字段date、hour、minute、adID、clickedCountList<Object[]> insertParametersList = new ArrayList<Object[]>();for(AdTrendStat insertRecord : inserting) {insertParametersList.add(new Object[] {insertRecord.get_date(),insertRecord.get_hour(),insertRecord.get_minute(),insertRecord.getAdID(),insertRecord.getClickedCount()});}jdbcWrapper.doBatch("INSERT INTO adclickedtrend VALUES(?, ?, ?, ?, ?)", insertParametersList);//表的字段date、hour、minute、adID、clickedCountList<Object[]> updateParametersList = new ArrayList<Object[]>();for(AdTrendStat updateRecord : updating) {updateParametersList.add(new Object[] {updateRecord.getClickedCount(),updateRecord.get_date(),updateRecord.get_hour(),updateRecord.get_minute(),updateRecord.getAdID()});}jdbcWrapper.doBatch("UPDATE adclickedtrend SET clickedCount = ? WHERE"+ " date =? AND hour = ? AND minute = ? AND AdID = ?", updateParametersList);} });return null;} });;/ Spark Streaming 执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于 接收应用程序本身或者Executor中的消息，/javassc.start();javassc.awaitTermination();javassc.close();}private static JavaStreamingContext createContext(String checkpointDirectory, SparkConf conf) {// If you do not see this printed, that means the StreamingContext has been loaded// from the new checkpointSystem.out.println("Creating new context");// Create the context with a 5 second batch sizeJavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(10));ssc.checkpoint(checkpointDirectory);return ssc;} }class JDBCWrapper {private static JDBCWrapper jdbcInstance = null;private static LinkedBlockingQueue<Connection> dbConnectionPool = new LinkedBlockingQueue<Connection>();static {try {Class.forName("com.mysql.jdbc.Driver");} catch (ClassNotFoundException e) {// TODO Auto-generated catch blocke.printStackTrace();} }public static JDBCWrapper getJDBCInstance() {if(jdbcInstance == null) {synchronized (JDBCWrapper.class) {if(jdbcInstance == null) {jdbcInstance = new JDBCWrapper();} }}return jdbcInstance; }private JDBCWrapper() {for(int i = 0; i < 10; i++){try {Connection conn = DriverManager.getConnection("jdbc:mysql://Master:3306/sparkstreaming","root", "root");dbConnectionPool.put(conn);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } }public synchronized Connection getConnection() {while(0 == dbConnectionPool.size()){try {Thread.sleep(20);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }return dbConnectionPool.poll();}public int[] doBatch(String sqlText, List<Object[]> paramsList){Connection conn = getConnection();PreparedStatement preparedStatement = null;int[] result = null;try {conn.setAutoCommit(false);preparedStatement = conn.prepareStatement(sqlText);for(Object[] parameters: paramsList) {for(int i = 0; i < parameters.length; i++){preparedStatement.setObject(i + 1, parameters[i]);} preparedStatement.addBatch();}result = preparedStatement.executeBatch();conn.commit();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }}return result; }public void doQuery(String sqlText, Object[] paramsList, ExecuteCallBack callback){Connection conn = getConnection();PreparedStatement preparedStatement = null;ResultSet result = null;try {preparedStatement = conn.prepareStatement(sqlText);for(int i = 0; i < paramsList.length; i++){preparedStatement.setObject(i + 1, paramsList[i]);} result = preparedStatement.executeQuery();try {callback.resultCallBack(result);} catch (Exception e) {// TODO Auto-generated catch blocke.printStackTrace();} } catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally {if(preparedStatement != null) {try {preparedStatement.close();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();} }if(conn != null) {try {dbConnectionPool.put(conn);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} }} }}interface ExecuteCallBack {void resultCallBack(ResultSet result) throws Exception;}class UserAdClicked {private String timestamp;private String ip;private String userID;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public String getUserID() {return userID;}public void setUserID(String userID) {this.userID = userID;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdClicked {private String timestamp;private String adID;private String province;private String city;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public String getCity() {return city;}public void setCity(String city) {this.city = city;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdProvinceTopN {private String timestamp;private String adID;private String province;private Long clickedCount;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp = timestamp;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public String getProvince() {return province;}public void setProvince(String province) {this.province = province;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendStat {private String _date;private String _hour;private String _minute;private String adID;private Long clickedCount;public String get_date() {return _date;}public void set_date(String _date) {this._date = _date;}public String get_hour() {return _hour;}public void set_hour(String _hour) {this._hour = _hour;}public String get_minute() {return _minute;}public void set_minute(String _minute) {this._minute = _minute;}public String getAdID() {return adID;}public void setAdID(String adID) {this.adID = adID;}public Long getClickedCount() {return clickedCount;}public void setClickedCount(Long clickedCount) {this.clickedCount = clickedCount;} }class AdTrendCountHistory{private Long clickedCountHistoryLong;public Long getClickedCountHistoryLong() {return clickedCountHistoryLong;}public void setClickedCountHistoryLong(Long clickedCountHistoryLong) {this.clickedCountHistoryLong = clickedCountHistoryLong;} } 本篇文章为转载内容。原文链接：https://blog.csdn.net/tom_8899_li/article/details/71194434。 该文由互联网用户投稿提供，文中观点代表作者本人意见，并不代表本站的立场。 作为信息平台，本站仅提供文章转载服务，并不拥有其所有权，也不对文章内容的真实性、准确性和合法性承担责任。 如发现本文存在侵权、违法、违规或事实不符的情况，请及时联系我们，我们将第一时间进行核实并删除相应内容。

...QL是一个开源的关系型数据库管理系统，广泛应用于网站和应用开发中。在本文的语境中，MySQL是用户进行数据存储、查询和管理的基础服务，当出现“Table database_name.table_name doesn t exist”错误时，意味着在MySQL中无法找到指定的数据库表。 权限 , 在数据库系统如MySQL中，权限是指用户对数据库或其中的对象（例如表）执行特定操作的能力。这些权限可以细分为创建、读取、更新、删除等不同类型，根据角色的不同而有所区别。在文章中提到，如果用户报告找不到数据库表，可能是由于没有足够的权限访问该表导致的。 数据库连接故障 , 在计算机网络和数据库技术领域中，数据库连接故障指的是客户端应用程序与MySQL服务器之间的通信出现问题，无法建立或维持有效的连接状态。可能的原因包括但不限于服务器未运行、网络中断、登录凭据错误（如用户名、密码不正确）、连接超时等。在本文的上下文中，数据库连接故障可能导致用户即使拥有正确的数据库和表名，也无法成功访问到所需的数据库表。

...窗口函数以支持复杂的数据分析，提升了安全性（如密码验证插件默认更改为caching_sha2_password），并增强了InnoDB存储引擎的性能。因此，在考虑升级MySQL版本时，开发者不仅需要关注当前运行环境下的版本兼容性，更要深入了解新版本功能是否能够提升应用效能或满足新的业务需求。 同时，MySQL的社区版与企业版之间也存在功能差异。企业用户在选择版本时需结合自身业务规模和技术支持需求来决定。例如，Oracle MySQL企业版提供了高级的集群解决方案、热备份工具及额外的监控选项，这些都是社区版不具备的功能。 此外，MySQL的替代品如PostgreSQL、MariaDB等数据库管理系统也在不断迭代发展，它们在特定场景下可能具备更优的性能或特性。因此，作为开发人员或IT管理员，在决定是否跟随MySQL最新版本更新，或者转向其他数据库系统时，应全面权衡技术选型、成本效益、团队技能储备等因素，并进行详尽的测试和评估。 总之，MySQL版本管理是持续的运维工作之一，理解不同版本的特点与变化趋势，结合实际应用场景制定合理的升级策略，将有助于提高系统的稳定性和应用的竞争力。

...种广泛使用的开源关系型数据库管理系统（RDBMS），由Oracle公司开发并维护。在本文的语境中，MySQL指的是用户需要在Windows操作系统下启动和管理的数据库服务。MySQL以其稳定、安全、性能优越和跨平台支持等特点，被众多网站、应用程序以及企业级系统选作数据存储解决方案。 命令行窗口 , 命令行窗口，又称为控制台或终端，是Windows操作系统中的一个界面程序，允许用户通过输入文本命令来与操作系统进行交互。在本文中，用户需通过命令行窗口执行特定的命令以启动MySQL服务器和连接到MySQL数据库，这包括更改目录至MySQL的bin目录，运行mysqld命令启动MySQL服务，以及使用mysql命令登录MySQL服务器等操作。 root用户 , 在MySQL数据库系统中，“root”是一个特殊的系统管理员账号，拥有对整个MySQL服务器及其所有数据库的最高权限。启动MySQL数据库后，用户通过命令行工具以root用户身份登录，可以执行创建数据库、修改用户权限、删除数据表等各种高级管理操作。在本文的步骤中，用户需要输入root用户的密码来验证身份，并进入MySQL的命令行界面进行后续管理任务。

...我们还可以进一步探讨数据库在游戏开发中的核心作用以及数据库优化对提升玩家体验的影响。近期，《游戏开发者》杂志的一篇文章揭示了某知名网络游戏通过优化数据库架构，成功减少了游戏内交易的延迟，显著提升了元宝充值、消耗等操作的实时性，从而提高了用户满意度和留存率。 同时，随着云计算和大数据技术的发展，许多游戏公司开始采用分布式数据库来应对高并发场景下的数据处理需求。例如，阿里云发布的最新解决方案中就详细介绍了如何借助云数据库实现动态扩容，有效支撑了大型网游在高峰期的海量元宝数值更新与查询请求。 此外，针对游戏经济系统的安全问题，也有专家提出应当强化数据库权限管理，采用加密传输技术和二次验证机制确保元宝等虚拟财产的安全存储与变更。最近一起因数据库漏洞导致的游戏元宝被盗事件，再次敲响了游戏数据安全的警钟，促使业界加大对数据库防护措施的研究和投入。 总的来说，从基本的MySQL操作到复杂的数据库架构设计与优化，再到数据安全防护，游戏开发过程中对于数据库技术的应用和探索是一个持续且深入的过程，它不仅影响着游戏功能的实现，更是关乎游戏生态健康与用户体验的关键因素。

...表 , 哈希表是一种数据结构，通过哈希函数将输入（键）转换为数组索引以实现快速访问。在Java的HashMap和HashSet中，哈希表被用于存储元素，通过计算键的哈希值来确定元素在数组中的位置，从而支持O(1)时间复杂度的查找、插入和删除操作。当多个键映射到同一个索引时，会发生哈希冲突，可以通过链地址法、开放寻址法等方式解决。 哈希函数 , 哈希函数是一种特殊的函数，它可以将任意长度的输入（通常是字符串或对象）转化为固定长度的输出，即哈希值。在Java的HashMap和HashSet中，哈希函数用于计算键的哈希码，该哈希码随后用于确定元素在哈希表中的存储位置。一个好的哈希函数应该尽可能均匀地分布不同输入的哈希值，以减少哈希冲突并提高性能。 O(1)时间复杂度 , 在计算机科学中，O(1)时间复杂度表示算法的运行时间与问题规模无关，即无论输入数据大小如何，其执行时间恒定。在讨论HashMap和HashSet时，查找、插入和删除操作的时间复杂度都是O(1)，意味着只要哈希函数设计合理且哈希表负载均衡，对集合进行这些基本操作所需的时间是常数级别的，不随集合内元素数量的增长而增长。但这需要假设哈希函数将键均匀分布到桶中，并且桶的数量足够大以保持较低的冲突率。

在Java开发领域中，PDF文档处理是一个常见且重要的需求。近期，随着Apache PDFBox库的不断更新与优化，它作为另一种强大的开源Java库，同样提供了高效、便捷的PDF与String互转功能，并逐渐受到开发者们的青睐。相较于iText，PDFBox在处理大量PDF文件时展现出了卓越的性能和更为灵活的功能支持。 例如，使用PDFBox将PDF转换为文本字符串，可以采用PDFTextStripper类，其API设计简洁易用，支持提取PDF中的富文本信息以及表格内容。而在创建PDF文件方面，PDFBox通过PDDocument、PDPage及PDPageContentStream等核心类，让开发者能够更加精细地控制PDF页面布局与内容填充，实现复杂报表、合同文档等多种类型的PDF生成。 值得注意的是，无论是iText还是PDFBox，都遵循了Apache 2.0开源协议，确保了开发者在商业项目中的自由使用。同时，它们均提供了详细的官方文档和丰富的社区资源，便于开发者深入学习和解决实际问题。 另外，随着云计算和微服务架构的发展，越来越多的服务提供商如Google Cloud、阿里云等也推出了基于RESTful API的云端PDF处理服务，使得开发者无需直接在本地应用中集成上述库，即可轻松完成PDF与字符串之间的转换任务，进一步提升了开发效率与系统的可扩展性。 综上所述，在面对PDF与String互转这一需求时，Java开发者可以根据具体场景选择合适的工具或服务，以适应快速变化的技术环境和业务需求。对于希望深入了解和掌握PDF处理技术的开发者来说，持续关注并研究这些最新的技术和解决方案，无疑将极大地提升自身在文档处理领域的专业能力。

...索更多高级定制和优化方案。例如，结合Vue.js的动态组件特性，可以设计更为灵活多样的展开内容区域，展示不同类型的嵌套数据或操作面板。此外，对于大数据量表格的性能优化也是值得关注的问题，通过懒加载、虚拟滚动等技术提高渲染效率。 近期，Element Plus作为Element-UI的下一代版本，在处理表格组件方面提供了更多的改进与优化，比如更强大的API支持、更流畅的交互体验以及对Tree Table结构的良好兼容性，为实现复杂表格交互提供了新思路。开发者们可以通过学习Element Plus的新特性，升级现有项目以提升用户体验并紧跟前端技术潮流。 同时，针对无障碍设计（Accessibility）的重要性日益凸显，如何确保el-table的展开/收起功能对键盘操作友好，符合WCAG 2.1标准，也成为了现代Web开发中的一个重要议题。通过对焦点管理、ARIA角色属性的合理设置，我们可以使所有用户，无论是否使用鼠标，都能顺畅地与具有展开/收起功能的数据表格进行交互。 综上所述，深入理解和掌握表格组件的扩展功能不仅有助于提升项目的用户体验，也有利于开发者关注前端领域最新技术和无障碍设计的发展趋势，从而打造出更加高效、易用且包容的Web应用。

...从Pod中自动收集、存储和转发日志数据，大大简化了大规模容器集群的日志管理工作流程。同时，众多开源项目如EFK（Elasticsearch、Fluentd、Kibana）栈或Loki等日志解决方案正与Kubernetes紧密集成，为用户提供实时检索、可视化分析及报警等功能，显著提升运维效率。 此外，在安全合规层面，针对容器日志的安全审计越来越受到重视。一些企业开始采用具有加密功能的日志传输协议，以及支持细粒度权限控制和长期存储的云端日志服务，确保容器产生的敏感信息能够得到妥善保护和合规留存。 总的来说，容器日志管理不仅涉及基础的操作技巧与工具配置，更需要紧跟行业发展潮流，掌握先进的日志架构设计与最佳实践，以适应日益复杂的应用场景和严苛的安全要求。通过不断优化日志系统，企业不仅能快速定位问题、提升应用服务质量，还能更好地满足业务连续性需求和监管政策规定。

...he Lucene：解决NoSuchDirectoryException找不到目录异常 引言 在进行大规模数据处理时，可能会遇到各种各样的问题，其中一个常见的问题是NoSuchDirectoryException找不到目录异常。这个异常通常会在我们尝试访问一个不存在的文件或目录时抛出。今天呢，咱们就来一起唠唠嗑，探讨一下如何借助这个超牛的搜索引擎工具——Apache Lucene，来把这个问题给妥妥地解决了哈！ 什么是Apache Lucene？ Apache Lucene是一个开源的全文搜索库，主要用于对文本进行索引和搜索。它支持多种语言，并且可以运行在多个操作系统上。Lucene的性能非常高，可以快速地对大量文本进行搜索。 NoSuchDirectoryException找不到目录异常 当我们在使用Lucene时，如果试图访问一个不存在的目录，就会抛出NoSuchDirectoryException异常。这是因为Lucene在启动的时候，得先建一个文件目录来存放索引和其它相关的那些文件啦。要是这个目录没影儿了，那就没法继续给Lucene走初始化流程了，这时候就得抛出个异常来提醒你。 例如，下面的代码尝试初始化一个名为test的Lucene实例： java Directory directory = FSDirectory.open(new File("test")); Analyzer analyzer = new StandardAnalyzer(); IndexWriterConfig config = new IndexWriterConfig(analyzer); IndexWriter writer = new IndexWriter(directory, config); 如果test目录不存在，这段代码就会抛出NoSuchDirectoryException异常。 解决NoSuchDirectoryException找不到目录异常的方法 为了解决这个问题，我们需要在初始化Lucene之前，先创建这个目录。我们可以使用Java的File类来创建这个目录。以下是一个示例： java try { File dir = new File("test"); if (!dir.exists()) { boolean success = dir.mkdir(); if (!success) { throw new RuntimeException("Failed to create directory."); } } Directory directory = FSDirectory.open(dir); Analyzer analyzer = new StandardAnalyzer(); IndexWriterConfig config = new IndexWriterConfig(analyzer); IndexWriter writer = new IndexWriter(directory, config); } catch (IOException e) { // Handle IOExceptions here. } 在这个示例中，我们首先检查test目录是否已经存在。如果不存在，我们就尝试创建它。如果创建失败，我们就抛出一个运行时异常。如果创建成功，我们就使用这个目录来初始化Lucene。 这样，即使test目录不存在，我们的代码也可以正常运行，并且能够创建一个新的目录。 结论 总的来说，NoSuchDirectoryException找不到目录异常是我们在使用Lucene时经常会遇到的问题。但是，只要我们掌握了正确的解决方案，就可以轻松地解决这个问题。在我们动手初始化Lucene之前，有个小窍门可以确保目录已经准备就绪，那就是用Java里的File类来亲手创建这个目录，这样一来，一切就能稳妥进行啦！这样一来，哪怕目录压根不存在，我们的代码也能稳稳地运行起来，并且顺手就把新的目录给创建了。

...了JSON作为轻量级数据交换格式的基础概念及其在JavaScript中的应用后，我们可进一步探索这一技术在现代Web开发及跨平台数据交互领域的最新动态与实践。 近年来，随着API经济的快速发展和微服务架构的广泛应用，JSON愈发成为主流的数据传输格式。例如，在GraphQL这一新兴的API查询语言中，JSON不仅被用作请求和响应的数据载体，还支持丰富的自定义类型系统，以满足日益复杂的应用场景需求。此外，诸如AJAX、RESTful API等技术也都深度依赖JSON进行前后端数据交互。 与此同时，考虑到性能优化和数据压缩的问题，业界也出现了对JSON的改进方案。比如，Facebook推出的Msgpack是一种二进制序列化格式，它在保持类似JSON语法简洁性的同时，显著提高了数据传输效率。另外，JSONB（Binary JSON）是PostgreSQL数据库为存储和检索JSON数据而提供的高效二进制格式。 不仅如此，针对JSON的安全性问题，开发者需关注如何有效验证和过滤JSON数据，防止注入攻击等安全风险。为此，一些库如ajv、 Joi等提供了严谨的数据模式验证功能，确保接收到的JSON数据符合预期结构和类型。 综上所述，深入理解和掌握JSON相关的最新技术和最佳实践，对于提升应用程序的数据处理能力、保障数据交互安全以及优化系统性能等方面具有重要价值。建议读者持续关注JSON及相关领域的发展趋势，并结合具体项目需求灵活运用各种解决方案。

...因为你可能不知道如何解决它。别担心，我们来一起探讨一下这个问题。 二、问题解析 首先，让我们来看看什么是磁盘分区识别错误。简单来说，当你打算把一个文件从一处搬到另一处，但这两个地方不在同一个磁盘分区上时，你的电脑操作系统就会犯迷糊，认不出磁盘分区，然后给你来个错误提示。这是因为不同的磁盘分区有不同的文件系统，如果你试图将文件从一种文件系统移动到另一种文件系统，操作系统就无法识别这个操作。 三、原因分析 那么，为什么我们在安装Kylin系统时会出现这种问题呢？这可能是由于以下几种原因： 1. 系统资源不足 如果你的计算机硬盘空间不足，系统可能无法正确地进行分区。 2. 文件系统不匹配 如果你试图将文件从一种文件系统移动到另一种文件系统，而这两个文件系统的版本不同，系统就可能出现识别错误。 3. 磁盘损坏 如果你的磁盘出现物理损坏，系统就可能无法正确地读取和写入数据。 四、解决方案 知道了问题的原因，我们就可以开始寻找解决问题的方法了。以下是一些常见的解决办法： 1. 扩展硬盘空间 如果你的硬盘空间不足，你可以尝试扩大硬盘的空间。这可以通过购买一个新的硬盘或者升级现有的硬盘来实现。 2. 更改文件系统 如果你试图将文件从一种文件系统移动到另一种文件系统，你可以尝试更改其中一个文件系统的版本。比如说，你要是想把文件从FAT32格式的盘挪到NTFS格式的盘，完全可以先把这个盘转换成NTFS格式，然后再进行文件搬家的操作。 3. 检查磁盘 如果你的磁盘出现物理损坏，你需要检查磁盘并修复或替换它。 五、实例演示 让我们来看一个具体的例子。假设你在安装Kylin系统时出现了磁盘分区识别错误。你可以按照以下步骤来解决问题： 1. 首先，检查你的硬盘空间。如果你的硬盘空间不足，你需要扩展硬盘空间。你可以通过购买一个新的硬盘或者升级现有的硬盘来实现。 2. 其次，检查你的文件系统。如果你想把文件从一个文件系统搬到另一个文件系统，那就得先瞧准了，这两个系统的版本得对得上号才行。你可以使用命令行工具来查看和更改文件系统的版本。例如，在Windows系统中，你可以使用fsutil fsinfo diskvolume信息来查看和更改文件系统的版本。 3. 最后，如果你的磁盘出现物理损坏，你需要检查磁盘并修复或替换它。你可以使用各种磁盘检测和修复工具来帮助你完成这个任务。 六、总结 总的来说，磁盘分区识别错误是一个比较常见的问题，但是只要你知道了它的原因，并且采取了正确的解决办法，你就能够成功地解决这个问题。记住了啊，不论你碰到啥困难、挑战，都要稳住心态，乐观面对，坚信自己肯定有办法把问题给解决了。别忘了，你可是个解决问题的小能手呢！

...了一些数字： javascript var numbers = [1, 2, 3, 4, 5]; 现在，我们需要把数组中的某个特定元素向前移动一位。例如，如果我们要将3这个元素向前移动一位，那么最终结果应该是： javascript [1, 3, 2, 4, 5] 3. 解决方案 3.1 分析 首先，我们需要找到这个元素在数组中的位置。然后，将它与前一个元素交换位置。这个过程听起来不难，但是实现起来需要考虑几个关键点。 3.2 实现步骤 1. 查找元素的位置 我们可以通过.indexOf()方法来获取元素的位置。 2. 判断边界条件 如果元素已经是第一个元素，那么就没有必要再往前移动了。 3. 交换元素位置 通过数组的splice方法来交换两个元素的位置。 让我们一步一步来看代码实现。 3.3 代码示例 javascript $(document).ready(function() { var numbers = [1, 2, 3, 4, 5]; // 找到元素的位置 var index = $.inArray(3, numbers); if (index !== -1 && index > 0) { // 判断是否是第一个元素 // 交换元素位置 var temp = numbers[index-1]; numbers[index-1] = numbers[index]; numbers[index] = temp; console.log(numbers); // 输出: [1, 3, 2, 4, 5] } else { console.log("元素已经在首位或者不存在"); } }); 这里，我们使用了jQuery的$.inArray()方法来查找元素的位置。如果我们找到了那个元素，并且它在数组里的位置不是第一个，那就把它和前面的那个元素换一下位置。 4. 进阶技巧 当然，这只是基本的实现方式。在实际开发中，你可能会碰到更棘手的情况，比如得反复挪动某个元素，或者它的位置总是变来变去，让你头大。这时候，你可以考虑封装一个函数来处理这种情况。 4.1 封装函数 javascript function moveElementForward(arr, element) { var index = $.inArray(element, arr); if (index !== -1 && index > 0) { var temp = arr[index-1]; arr[index-1] = arr[index]; arr[index] = temp; } return arr; } $(document).ready(function() { var numbers = [1, 2, 3, 4, 5]; console.log(moveElementForward(numbers, 3)); // 输出: [1, 3, 2, 4, 5] }); 这样，每次调用moveElementForward()函数时，就可以方便地将指定元素向前移动一位，而不需要重复编写相同的代码。 5. 结语 通过这次的技术分享，我们不仅学习了如何使用jQuery来处理数组中的元素移动，还了解了一些进阶的编程技巧。编程不仅仅是技术上的挑战，更是一种思维方式的锻炼。希望这篇东西能给你点灵感，在以后的项目里玩转jQuery就像吃糖一样简单。 最后，如果你有任何疑问或者更好的解决方案，请随时留言交流。编程之路，我们一起前行！

...业界使用最广泛的关系型数据库管理系统（RDBMS），其具有高效能、扩展性好、方便易用等优点。随着互联网应用的普及，数据量越来越大，单机MySQL已经难以满足业务需求，因此MySQL的分散式就成为了热议焦点。那么，如何领会MySQL的分散式呢？ 首先，我们需要了解分散式系统的理念。分散式系统是由若干台独立计算机通过网络进行协同与通讯，实现信息共享和协作作业的一个系统。在分散式系统中，不同计算机处理不同的任务，相互之间相互协作、协调，完成整个系统的功能。现在，将这个理念应用到MySQL中，我们可以说MySQL的分散式就是由若干台计算机组成的一个系统，可以分担MySQL的读写压力，提高CPU、内存等硬件资源使用率，从而达到更高的吞吐量、更高的并发性能。 MySQL的分散式，主要有两种实现方式： 1. MySQL Proxy：MySQL Proxy是一个轻量级的可插入的中间件，用于分发数据库负载，并实现复制和高可用性（HA）。它可以处理大量的并发连接和查询，并能够将这些请求转发到不同的MySQL数据库上。MySQL Proxy提供了可编程性，使其能够扩展和自定义，以适应不同的需求。 2. MySQL Cluster：MySQL Cluster是一个基于InnoDB存储引擎的面向事务的分散式数据库系统。它使用自己的数据节点和数据复制技术，实现平滑的水平扩展，提供高可用性和高可扩展性，支持分散式事务和分区表。MySQL Cluster尤其适合处理实时的在线业务应用，如电信、金融、电子商务等。 总之，MySQL的分散式是现代互联网应用的必备技术之一，它可以提高MySQL的可扩展性和高效能，同时也增加了系统的稳定性和可用性。对于需要处理大量读写请求和海量数据存储的应用，MySQL的分散式是一个非常好的解决方案。

...字化时代，MySQL数据库的重要性不言而喻。近日（请根据实际日期填充），Oracle公司发布了MySQL 8.0的最新版本，引入了诸多性能优化和新特性，如窗口函数、原子DDL操作以及改进的安全模块等，进一步提升了MySQL在大规模数据处理与安全防护上的能力。 针对日益严峻的数据安全问题，InfoWorld网站近期发布了一篇深度分析文章，探讨了如何通过实施严格的访问控制策略、加密敏感数据及定期审计来强化MySQL数据库的安全性。此外，文中还介绍了业界最新的数据保护法规GDPR对数据库管理的影响，提醒用户在使用MySQL时需遵循合规要求。 同时，鉴于云服务的普及，Amazon RDS for MySQL作为一种托管型数据库服务备受关注。AWS官方博客分享了关于如何高效迁移本地MySQL数据库至RDS，并实现无缝备份与恢复的实战经验，为众多寻求上云解决方案的企业提供了宝贵参考。 不仅如此，对于希望深入理解MySQL内部机制的开发者，Stack Overflow上有资深专家撰写了系列教程，详尽解析了InnoDB存储引擎的工作原理，以及SQL查询优化技巧，帮助读者提升数据库设计与运维水平。 总之，在掌握MySQL基本使用的基础上，持续跟进技术发展动态，深入了解并实践高级功能与安全管理措施，是确保MySQL数据库在各类型应用程序中稳定高效运行的关键。

一、引言 在大数据处理中，数据一致性是一个至关重要的问题。无论是存东西、找信息还是分析数据，数据一致性这玩意儿都直接关系到结果靠不靠谱、准不准。在这篇文章里，我们打算好好聊聊DorisDB在应对数据文件重复或者发生冲突时，可能会遇到的一些头疼问题，并且还会送上咱们精心准备的解决大招~ 二、数据文件重复与冲突的影响 1. 数据冗余 当同一个数据被多个文件重复存储时，就会出现数据冗余。这不仅浪费了存储空间，还可能导致数据更新时出现问题。 2. 数据一致性 如果数据文件之间存在冲突，那么可能会导致数据的一致性受到影响。比如，假设有两个文件同时对一个数据进行修改，如果没有靠谱的冲突解决办法，那么最后的数据结果就可能会乱套，一致性就无法得到保障啦。 三、使用DorisDB处理数据文件重复或冲突 1. 使用唯一索引 在DorisDB中，我们可以为表中的每个字段设置唯一的索引，以此来防止数据文件的重复。例如： java alter table my_table add unique index idx_my_field (my_field); 2. 使用事务 如果存在多个文件需要对同一份数据进行操作的情况，可以使用DorisDB的事务功能来确保数据的一致性。例如： java begin; update my_table set my_field = 1 where id = 1; commit; 四、结论 虽然数据文件的重复或冲突可能会给DorisDB带来一些挑战，但通过正确的使用DorisDB的功能，我们完全可以有效地管理和处理这些问题。在接下来的工作里，我们还要继续钻研和搜寻更多给力的方法，目标是让DorisDB在应对数据文件重复或冲突这类问题时，能够更高效、更稳当地运转起来，就像跑车换上了更强悍的引擎一样。

在深入理解Java开发中的前后台交互原理和技术实现后，我们可以进一步探索现代Web开发领域的一些最新趋势和最佳实践。例如，随着前后端分离架构的普及，RESTful API设计原则已成为主流，它强调接口的简洁性、可读性和可扩展性，使得前端与后端通过HTTP请求进行数据交互更为高效和规范。 另外，近年来，微服务架构逐渐成为大型分布式系统设计的首选方案，Spring Boot和Spring Cloud等框架提供了完善的微服务解决方案，它们不仅简化了后台服务的构建过程，而且强化了不同服务之间的交互能力，确保前后台之间能够更加灵活地进行数据通信。 同时，前端技术也在快速发展，如React、Vue.js等现代JavaScript框架，它们通过组件化、状态管理等机制，更好地实现了与后台API的数据绑定和实时更新，提升了用户体验，并且有助于前后端团队并行开发，提高项目整体效率。 此外，在数据传输格式上，JSON虽仍为主流，但诸如GraphQL这样的查询语言也开始崭露头角，它允许客户端指定需要获取的具体数据字段，从而减少网络传输量，提高响应速度。 综上所述，无论是在架构设计、开发框架选择，还是在数据交换格式方面，Java Web开发中的前后台交互都在持续演进和完善中，开发者应紧跟技术发展趋势，结合实际业务需求，以实现更高效、更稳定的前后台交互体验。

数据同步工具 , 数据同步工具是一种软件应用，主要用于在不同的数据库系统、文件系统或其他数据存储之间传输和更新数据。在本文语境中，Datax即是一款开源的数据同步工具，它支持多种数据库类型，可实现高效、稳定的数据迁移和清洗任务。 数据库权限 , 在数据库管理系统中，权限是指用户或角色对数据库对象（如表、视图、存储过程等）可以执行的操作范围。例如，INSERT权限允许用户向表中插入新记录，UPDATE权限允许修改已有记录，而DELETE权限则允许删除记录。当使用Datax连接源数据库时，若出现授权失败的问题，可能是由于指定的数据库用户名没有足够的权限来执行所需操作。 防火墙规则 , 防火墙规则是网络防火墙为了控制进出特定网络的数据包流量而制定的一系列策略。这些规则根据源地址、目标地址、端口号以及协议类型等因素定义了允许或拒绝哪些网络通信。在本文情境下，如果数据库防火墙阻止了Datax的连接请求，就需要调整防火墙规则，允许来自运行Datax服务器的IP地址进行连接，确保数据同步任务能够正常进行。

...ySQL作为开源关系型数据库管理系统的基础操作后，进一步的“延伸阅读”可以聚焦于以下几个方面： 首先，针对MySQL的最新发展动态，近期Oracle公司发布了MySQL 8.0版本，引入了一系列性能优化和新特性，如窗口函数、原子DDL操作以及增强的安全功能（如caching_sha2_password认证插件），这些改进对于系统数据存储与管理的安全性和效率都带来了显著提升。 其次，随着云服务的发展，各大云服务商如AWS、阿里云、腾讯云等均提供了MySQL托管服务，用户无需关心底层硬件维护与软件升级，只需关注数据模型设计和SQL查询优化，大大降低了数据库运维门槛。例如，AWS RDS MySQL服务提供了一键备份恢复、读写分离、自动扩展等功能，为系统数据的高效管理和高可用性提供了有力支持。 再者，深入探讨MySQL在大数据处理领域的应用也不容忽视。虽然MySQL传统上主要用于OLTP在线交易处理场景，但在结合Hadoop、Spark等大数据框架后，也能够实现大规模数据分析和处理。比如使用Apache Sqoop工具将MySQL数据导入HDFS，或通过JDBC连接Spark SQL对MySQL数据进行复杂分析。 此外，对于系统安全性的考虑，如何有效防止SQL注入、实施权限管理以及加密敏感数据也是MySQL使用者需要关注的重点。MySQL自带的多层访问控制机制及密码加密策略可确保数据安全性，同时，业界还推荐遵循OWASP SQL注入防护指南来编写安全的SQL查询语句。 总之，在实际工作中，熟练掌握MySQL并结合最新的技术趋势与最佳实践，将有助于构建更为稳定、高效且安全的系统数据存储解决方案。