- 浏览: 1335979 次
- 性别:
- 来自: 成都
文章分类
- 全部博客 (471)
- 原创文章 (4)
- Database (84)
- J2SE (63)
- Web (26)
- Javascript (30)
- Lucene (11)
- os (13)
- 算法 (8)
- Webservice (1)
- Open projects (18)
- Hibernate (18)
- Spring (15)
- Css (2)
- J2ee (2)
- 综合技术 (18)
- 安全管理 (13)
- PatternsInJava (27)
- NIO (5)
- Ibatis (2)
- 书籍收藏 (1)
- quartz (7)
- 并发编程 (15)
- oracle问题 (2)
- ios (60)
- coco2d-iphone (3)
- C++ (6)
- Zookeeper (2)
- golang (4)
- animation (2)
- android (1)
最新评论
-
dandingge123:
【引用】限制UITextField输入长度的方法 -
qja:
...
对List顺序,逆序,随机排列实例代码 -
安静听歌:
现在在搞这个,,,,,哎~头都大了,,,又freemarker ...
通用大型网站页面静态化解决方案(一) -
springdata-jpa:
java quartz定时任务demo教程源代码下载,地址:h ...
Quartz 配置参考 -
马清天:
[b][/b][list][*]引用[u][/u][/list ...
通用大型网站页面静态化解决方案(一)
ConcurrentHashMap是Java 5中支持高并发、高吞吐量的线程安全HashMap实现。在这之前我对ConcurrentHashMap只有一些肤浅的理解,仅知道它采用了多个锁,大概也足够了。但是在经过一次惨痛的面试经历之后,我觉得必须深入研究它的实现。面试中被问到读是否要加锁,因为读写会发生冲突,我说必须要加锁,我和面试官也因此发生了冲突,结果可想而知。还是闲话少说,通过仔细阅读源代码,现在总算理解ConcurrentHashMap实现机制了,其实现之精巧,令人叹服,与大家共享之。
实现原理
锁分离 (Lock Stripping)
ConcurrentHashMap允许多个修改操作并发进行,其关键在于使用了锁分离技术。它使用了多个锁来控制对hash表的不同部分进行的修改。ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分,每个段其实就是一个小的hash table,它们有自己的锁。只要多个修改操作发生在不同的段上,它们就可以并发进行。
有些方法需要跨段,比如size()和containsValue(),它们可能需要锁定整个表而而不仅仅是某个段,这需要按顺序锁定所有段,操作完毕后,又按顺序释放所有段的锁。这里“按顺序”是很重要的,否则极有可能出现死锁,在ConcurrentHashMap内部,段数组是final的,并且其成员变量实际上也是final的,但是,仅仅是将数组声明为final的并不保证数组成员也是final的,这需要实现上的保证。这可以确保不会出现死锁,因为获得锁的顺序是固定的。不变性是多线程编程占有很重要的地位,下面还要谈到。
- /**
- * The segments, each of which is a specialized hash table
- */
- final Segment<K,V>[] segments;
不变(Immutable)和易变(Volatile)
ConcurrentHashMap完全允许多个读操作并发进行,读操作并不需要加锁。如果使用传统的技术,如HashMap中的实现,如果允许可以在hash链的中间添加或删除元素,读操作不加锁将得到不一致的数据。ConcurrentHashMap实现技术是保证HashEntry几乎是不可变的。HashEntry代表每个hash链中的一个节点,其结构如下所示:
- static final class HashEntry<K,V> {
- final K key;
- final int hash;
- volatile V value;
- final HashEntry<K,V> next;
- }
可以看到除了value不是final的,其它值都是final的,这意味着不能从hash链的中间或尾部添加或删除节点,因为这需要修改next引用值,所有的节点的修改只能从头部开始。对于put操作,可以一律添加到Hash链的头部。但是对于remove操作,可能需要从中间删除一个节点,这就需要将要删除节点的前面所有节点整个复制一遍,最后一个节点指向要删除结点的下一个结点。这在讲解删除操作时还会详述。为了确保读操作能够看到最新的值,将value设置成volatile,这避免了加锁。
其它
为了加快定位段以及段中hash槽的速度,每个段hash槽的的个数都是2^n,这使得通过位运算就可以定位段和段中hash槽的位置。当并发级别为默认值16时,也就是段的个数,hash值的高4位决定分配在哪个段中。但是我们也不要忘记《算法导论》给我们的教训:hash槽的的个数不应该是2^n,这可能导致hash槽分配不均,这需要对hash值重新再hash一次。(这段似乎有点多余了 )
这是重新hash的算法,还比较复杂,我也懒得去理解了。
- private static int hash(int h) {
- // Spread bits to regularize both segment and index locations,
- // using variant of single-word Wang/Jenkins hash.
- h += (h << 15) ^ 0xffffcd7d;
- h ^= (h >>> 10);
- h += (h << 3);
- h ^= (h >>> 6);
- h += (h << 2) + (h << 14);
- return h ^ (h >>> 16);
- }
这是定位段的方法:
- final Segment<K,V> segmentFor(int hash) {
- return segments[(hash >>> segmentShift) & segmentMask];
- }
数据结构
关于Hash表的基础数据结构,这里不想做过多的探讨。Hash表的一个很重要方面就是如何解决hash冲突,ConcurrentHashMap和HashMap使用相同的方式,都是将hash值相同的节点放在一个hash链中。与HashMap不同的是,ConcurrentHashMap使用多个子Hash表,也就是段(Segment)。下面是ConcurrentHashMap的数据成员:
- public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
- implements ConcurrentMap<K, V>, Serializable {
- /**
- * Mask value for indexing into segments. The upper bits of a
- * key's hash code are used to choose the segment.
- */
- final int segmentMask;
- /**
- * Shift value for indexing within segments.
- */
- final int segmentShift;
- /**
- * The segments, each of which is a specialized hash table
- */
- final Segment<K,V>[] segments;
- }
所有的成员都是final的,其中segmentMask和segmentShift主要是为了定位段,参见上面的segmentFor方法。
每个Segment相当于一个子Hash表,它的数据成员如下:
- static final class Segment<K,V> extends ReentrantLock implements Serializable {
- private static final long serialVersionUID = 2249069246763182397L;
- /**
- * The number of elements in this segment's region.
- */
- transient volatile int count;
- /**
- * Number of updates that alter the size of the table. This is
- * used during bulk-read methods to make sure they see a
- * consistent snapshot: If modCounts change during a traversal
- * of segments computing size or checking containsValue, then
- * we might have an inconsistent view of state so (usually)
- * must retry.
- */
- transient int modCount;
- /**
- * The table is rehashed when its size exceeds this threshold.
- * (The value of this field is always <tt>(int)(capacity *
- * loadFactor)</tt>.)
- */
- transient int threshold;
- /**
- * The per-segment table.
- */
- transient volatile HashEntry<K,V>[] table;
- /**
- * The load factor for the hash table. Even though this value
- * is same for all segments, it is replicated to avoid needing
- * links to outer object.
- * @serial
- */
- final float loadFactor;
- }
count用来统计该段数据的个数,它是volatile,它用来协调修改和读取操作,以保证读取操作能够读取到几乎最新的修改。协调方式是这样的,每次修改操作做了结构上的改变,如增加/删除节点(修改节点的值不算结构上的改变),都要写count值,每次读取操作开始都要读取count的值。这利用了Java 5中对volatile语义的增强,对同一个volatile变量的写和读存在happens-before关系。modCount统计段结构改变的次数,主要是为了检测对多个段进行遍历过程中某个段是否发生改变,在讲述跨段操作时会还会详述。threashold用来表示需要进行rehash的界限值。table数组存储段中节点,每个数组元素是个hash链,用HashEntry表示。table也是volatile,这使得能够读取到最新的table值而不需要同步。loadFactor表示负载因子。
实现细节
修改操作
先来看下删除操作remove(key)。
- public V remove(Object key) {
- hash = hash(key.hashCode());
- return segmentFor(hash).remove(key, hash, null);
- }
整个操作是先定位到段,然后委托给段的remove操作。当多个删除操作并发进行时,只要它们所在的段不相同,它们就可以同时进行。下面是Segment的remove方法实现:
- V remove(Object key, int hash, Object value) {
- lock();
- try {
- int c = count - 1;
- HashEntry<K,V>[] tab = table;
- int index = hash & (tab.length - 1);
- HashEntry<K,V> first = tab[index];
- HashEntry<K,V> e = first;
- while (e != null && (e.hash != hash || !key.equals(e.key)))
- e = e.next;
- V oldValue = null;
- if (e != null) {
- V v = e.value;
- if (value == null || value.equals(v)) {
- oldValue = v;
- // All entries following removed node can stay
- // in list, but all preceding ones need to be
- // cloned.
- ++modCount;
- HashEntry<K,V> newFirst = e.next;
- for (HashEntry<K,V> p = first; p != e; p = p.next)
- newFirst = new HashEntry<K,V>(p.key, p.hash,
- newFirst, p.value);
- tab[index] = newFirst;
- count = c; // write-volatile
- }
- }
- return oldValue;
- } finally {
- unlock();
- }
- }
整个操作是在持有段锁的情况下执行的,空白行之前的行主要是定位到要删除的节点e。接下来,如果不存在这个节点就直接返回null,否则就要将e前面的结点复制一遍,尾结点指向e的下一个结点。e后面的结点不需要复制,它们可以重用。下面是个示意图,我直接从这个网站 上复制的(画这样的图实在是太麻烦了,如果哪位有好的画图工具,可以推荐一下)。
删除元素之前:
删除元素3之后:
第二个图其实有点问题,复制的结点中应该是值为2的结点在前面,值为1的结点在后面,也就是刚好和原来结点顺序相反,还好这不影响我们的讨论。
整个remove实现并不复杂,但是需要注意如下几点。第一,当要删除的结点存在时,删除的最后一步操作要将count的值减一。这必须是最后一步操作,否则读取操作可能看不到之前对段所做的结构性修改。第二,remove执行的开始就将table赋给一个局部变量tab,这是因为table是volatile变量,读写volatile变量的开销很大。编译器也不能对volatile变量的读写做任何优化,直接多次访问非volatile实例变量没有多大影响,编译器会做相应优化。
接下来看put操作,同样地put操作也是委托给段的put方法。下面是段的put方法:
- V put(K key, int hash, V value, boolean onlyIfAbsent) {
- lock();
- try {
- int c = count;
- if (c++ > threshold) // ensure capacity
- rehash();
- HashEntry<K,V>[] tab = table;
- int index = hash & (tab.length - 1);
- HashEntry<K,V> first = tab[index];
- HashEntry<K,V> e = first;
- while (e != null && (e.hash != hash || !key.equals(e.key)))
- e = e.next;
- V oldValue;
- if (e != null) {
- oldValue = e.value;
- if (!onlyIfAbsent)
- e.value = value;
- }
- else {
- oldValue = null;
- ++modCount;
- tab[index] = new HashEntry<K,V>(key, hash, first, value);
- count = c; // write-volatile
- }
- return oldValue;
- } finally {
- unlock();
- }
- }
该方法也是在持有段锁的情况下执行的,首先判断是否需要rehash,需要就先rehash。接着是找是否存在同样一个key的结点,如果存在就直接替换这个结点的值。否则创建一个新的结点并添加到hash链的头部,这时一定要修改modCount和count的值,同样修改count的值一定要放在最后一步。put方法调用了rehash方法,reash方法实现得也很精巧,主要利用了table的大小为2^n,这里就不介绍了。
修改操作还有putAll和replace。putAll就是多次调用put方法,没什么好说的。replace甚至不用做结构上的更改,实现要比put和delete要简单得多,理解了put和delete,理解replace就不在话下了,这里也不介绍了。
获取操作
首先看下get操作,同样ConcurrentHashMap的get操作是直接委托给Segment的get方法,直接看Segment的get方法:
- V get(Object key, int hash) {
- if (count != 0) { // read-volatile
- HashEntry<K,V> e = getFirst(hash);
- while (e != null) {
- if (e.hash == hash && key.equals(e.key)) {
- V v = e.value;
- if (v != null)
- return v;
- return readValueUnderLock(e); // recheck
- }
- e = e.next;
- }
- }
- return null;
- }
get操作不需要锁。第一步是访问count变量,这是一个volatile变量,由于所有的修改操作在进行结构修改时都会在最后一步写count变量,通过这种机制保证get操作能够得到几乎最新的结构更新。对于非结构更新,也就是结点值的改变,由于HashEntry的value变量是volatile的,也能保证读取到最新的值。接下来就是对hash链进行遍历找到要获取的结点,如果没有找到,直接访回null。对hash链进行遍历不需要加锁的原因在于链指针next是final的。但是头指针却不是final的,这是通过getFirst(hash)方法返回,也就是存在table数组中的值。这使得getFirst(hash)可能返回过时的头结点,例如,当执行get方法时,刚执行完getFirst(hash)之后,另一个线程执行了删除操作并更新头结点,这就导致get方法中返回的头结点不是最新的。这是可以允许,通过对count变量的协调机制,get能读取到几乎最新的数据,虽然可能不是最新的。要得到最新的数据,只有采用完全的同步。
最后,如果找到了所求的结点,判断它的值如果非空就直接返回,否则在有锁的状态下再读一次。这似乎有些费解,理论上结点的值不可能为空,这是因为put的时候就进行了判断,如果为空就要抛NullPointerException。空值的唯一源头就是HashEntry中的默认值,因为HashEntry中的value不是final的,非同步读取有可能读取到空值。仔细看下put操作的语句:tab[index] = new HashEntry<K,V>(key, hash, first, value),在这条语句中,HashEntry构造函数中对value的赋值以及对tab[index]的赋值可能被重新排序,这就可能导致结点的值为空。这种情况应当很罕见,一旦发生这种情况,ConcurrentHashMap采取的方式是在持有锁的情况下再读一遍,这能够保证读到最新的值,并且一定不会为空值。
- V readValueUnderLock(HashEntry<K,V> e) {
- lock();
- try {
- return e.value;
- } finally {
- unlock();
- }
- }
另一个操作是containsKey,这个实现就要简单得多了,因为它不需要读取值:
- boolean containsKey(Object key, int hash) {
- if (count != 0) { // read-volatile
- HashEntry<K,V> e = getFirst(hash);
- while (e != null) {
- if (e.hash == hash && key.equals(e.key))
- return true;
- e = e.next;
- }
- }
- return false;
- }
跨段操作
有些操作需要涉及到多个段,比如说size(), containsValaue()。先来看下size()方法:
- public int size() {
- final Segment<K,V>[] segments = this.segments;
- long sum = 0;
- long check = 0;
- int[] mc = new int[segments.length];
- // Try a few times to get accurate count. On failure due to
- // continuous async changes in table, resort to locking.
- for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
- check = 0;
- sum = 0;
- int mcsum = 0;
- for (int i = 0; i < segments.length; ++i) {
- sum += segments[i].count;
- mcsum += mc[i] = segments[i].modCount;
- }
- if (mcsum != 0) {
- for (int i = 0; i < segments.length; ++i) {
- check += segments[i].count;
- if (mc[i] != segments[i].modCount) {
- check = -1; // force retry
- break;
- }
- }
- }
- if (check == sum)
- break;
- }
- if (check != sum) { // Resort to locking all segments
- sum = 0;
- for (int i = 0; i < segments.length; ++i)
- segments[i].lock();
- for (int i = 0; i < segments.length; ++i)
- sum += segments[i].count;
- for (int i = 0; i < segments.length; ++i)
- segments[i].unlock();
- }
- if (sum > Integer.MAX_VALUE)
- return Integer.MAX_VALUE;
- else
- return (int)sum;
- }
size方法主要思路是先在没有锁的情况下对所有段大小求和,如果不能成功(这是因为遍历过程中可能有其它线程正在对已经遍历过的段进行结构性更新),最多执行RETRIES_BEFORE_LOCK次,如果还不成功就在持有所有段锁的情况下再对所有段大小求和。在没有锁的情况下主要是利用Segment中的modCount进行检测,在遍历过程中保存每个Segment的modCount,遍历完成之后再检测每个Segment的modCount有没有改变,如果有改变表示有其它线程正在对Segment进行结构性并发更新,需要重新计算。
其实这种方式是存在问题的,在第一个内层for循环中,在这两条语句sum += segments[i].count; mcsum += mc[i] = segments[i].modCount;之间,其它线程可能正在对Segment进行结构性的修改,导致segments[i].count和segments[i].modCount读取的数据并不一致。这可能使size()方法返回任何时候都不曾存在的大小,很奇怪javadoc居然没有明确标出这一点,可能是因为这个时间窗口太小了吧。size()的实现还有一点需要注意,必须要先segments[i].count,才能segments[i].modCount,这是因为segment[i].count是对volatile变量的访问,接下来segments[i].modCount才能得到几乎最新的值(前面我已经说了为什么只是“几乎”了)。这点在containsValue方法中得到了淋漓尽致的展现:
这种迭代方式的主要效果是不会抛出ConcurrentModificationException。一旦获取到下一个段的table,也就意味着这个段的头结点在迭代过程中就确定了,在迭代过程中就不能反映对这个段节点并发的删除和添加,对于节点的更新是能够反映的,因为节点的值是一个volatile变量。
结束语
ConcurrentHashMap是一个支持高并发的高性能的HashMap实现,它支持完全并发的读以及一定程度并发的写。ConcurrentHashMap的实现也是很精巧,充分利用了最新的JMM规范,值得学习,却不值得模仿。最后由于本人水平有限,对大师的作品难免有误解,如果存在,还望大牛们不吝指出。
如果要可靠的话,还是要使用Collections.synchronizedMap(Map map)方法,利用的synchronized机制,保证数据的一致性,然而ConcurrentHashMap只是保证Happens-before关系。
参考文章:
http://www.ibm.com/developerworks/java/library/j-jtp08223/,这个是讨论的是Doug Lea's util.concurrent包中的ConcurrentHashMap的实现,不过大致思想是一致的。
http://floatingpoint.tinou.com/2008/09/performance-optimization-in-concurrenthashmap.html
发表评论
-
J2ME 的优化措施
2012-07-11 02:15 1177常见的性能瓶颈:1.文件的读写和网络的操作2.CPU3.内存4 ... -
JAVA缓存 - Cache类
2011-09-28 11:04 4618JAVA缓存有两种: 一、文件缓存,是指把数据存储在磁盘上 ... -
Java注释(Annotation)详解
2011-03-31 18:22 2402注释:在一个类中可以有多个,类似于一个类可以有多个实例; ... -
Java去除字符串中的空格、回车、换行符、制表符
2011-02-15 11:06 3557本方法采用的是java的正则表达式 import jav ... -
线程池编程
2011-01-23 21:02 1105线程池编程 -
设计模式:Java语言中的应用
2011-01-23 20:59 1325设计模式:Java语言中的应用 -
java用了中间缓存变量
2011-01-23 18:37 2402Java语法、语言特性与机制 2011-01-22 19:1 ... -
Java集合总结
2011-01-19 11:16 3824Set:不区分元素的顺序,不允许出现重复的值list:区分元素 ... -
java.util.current的主要方法简介
2011-01-08 22:18 3436JDK5中增加了Doug Lea的并发库,这一引进给java线 ... -
线程池
2011-01-08 21:03 2071对每个客户都分配一个新的工作线程。当工作线程与客户通信结束,这 ... -
SimpleDateFormat
2010-12-10 16:11 1365With SimpleDateFo ... -
使用 XStream 在 JavaBean 与 XML/JSON 之间相互转换
2010-12-10 14:24 1562XML 和 JSON 是当今常用的两种数据描述与传输的格式,特 ... -
Linux上JDK的 安装和卸载 详细 (转)
2010-11-27 15:48 1927安装 1. 去http://java.sun.com ... -
java 路径问题
2010-11-26 16:21 13041.JSP/Servlet中获得当前应用的相对路径和绝对路径 ... -
对synchronized(this)的一些理解
2010-11-25 23:41 1027一、当两个并发线程访问同一个对象object中的这个synch ... -
Externalizable
2010-11-20 22:41 1363被Serializable接口声明的类的对象的内容都将被序列化 ... -
Java实现文件拷贝的4种方法(转)
2010-11-15 12:33 1251... -
Java内省和反射机制三步曲之 - 内省
2010-11-11 12:15 1344经过多方面的资料搜集整理,写下了这篇文章,本文主要讲解java ... -
java内省机制
2010-11-11 11:10 2400反射 (Reflection):反射就是让你可以通 ... -
Design Pattern: Thread-Specific Storage 模式
2010-11-11 00:00 1215无论如何,要编写一个多执行绪安全(thread-safe)的程 ...
相关推荐
ConcurrentHashMap是Java中提供的一种高效、线程安全的哈希表实现。与传统的基于synchronized关键字实现线程安全的HashTable相比,ConcurrentHashMap通过采用锁分段技术显著提高了并发性能。本文将深入探讨...
这个问题是由ConcurrentHashMap的实现细节所引起的。 ConcurrentHashMap是一个高效的哈希表实现,它可以在高并发环境下提供高性能的数据存储和检索。但是,在JDK1.8中,ConcurrentHashMap的实现存在一个严重的bug,...
在面试中,ConcurrentHashMap的底层原理、put方法的实现细节都是高频考点。本文将对ConcurrentHashMap#put方法的源码进行详细分析,从而帮助读者更好地理解ConcurrentHashMap的工作机理。 一、ConcurrentHashMap的...
而ConcurrentHashMap是线程安全的HashMap实现,它在Java 7中采用了分段锁(Segment)的设计,每个Segment实际上是一个小型的HashMap,通过锁来确保并发安全。put过程包括: 1. 确保Segment初始化,如果需要则创建新...
在Java面试中,经常会问到关于数据结构如HashTable和ConcurrentHashMap的细节,以及它们在并发编程中的使用。 最后,文档中出现了诸如“2399”、“1328”、“2645”、“2633”等数字,很可能是引用了一些代码片段或...
Java Core Sprout:一个萌芽阶段的Java核心知识库。...ConcurrentHashMap 的实现原理 如何优雅地使用和理解线程池 深入理解线程通信 一个线程召集的诡异事件 线程池中你不可错过的一些细节 『ARM包入坑指北』之队列
接下来,我们将详细探讨此程序的设计理念、关键技术和实现细节。 #### 二、关键技术点 1. **ConcurrentHashMap的应用**: - 在Java中,`ConcurrentHashMap`是一种线程安全的哈希表,适用于多线程环境下的并发访问...
#### 三、ConcurrentHashMap 的实现细节 **1. ConcurrentHashMap 结构** - `ConcurrentHashMap` 由一个 Segment 数组和多个哈希表组成。Segment 是一种可重入锁,每个 Segment 负责维护一部分哈希表。 - 每个 ...
本文将深入探讨`ConcurrentHashSet`的源码,解析其设计原理和实现细节。 首先,`ConcurrentHashSet`的核心是基于` ConcurrentHashMap `(并发哈希映射)来实现的,这使得它在多线程环境下具有高效性和线程安全性。`...
`Java中的几个HashMap ConcurrentHashMap实现分析Java开发Java经验技巧共4页.pdf.zip`这个压缩包文件很可能包含了一些深入的分析和实践案例,可以帮助你更好地理解和运用这些数据结构。在实践中不断探索和总结,是...
通过对源码的阅读和分析,我们可以更深入地理解LRU缓存的工作原理和具体实现细节。为了进一步学习和应用,你可以尝试阅读源码,理解每个类和方法的作用,甚至修改和扩展这个实现以满足特定需求。
- **编译器与运行时**:如`com.sun.*`和`sun.*`,虽然这些包不建议直接使用,但它们包含了JVM和编译器的相关实现细节。 **压缩包子文件的文件名称列表**:这些文件名暗示了源码的组织结构,如`launcher`可能包含...
常用集合 数组列表/向量 链表 哈希映射 ...ConcurrentHashMap 的实现原理 如何优雅地使用和理解线程池 深入理解线程通信 一个线程召集的诡异事件 线程池中你不可错过的一些细节 『ARM包入坑指北』之队列
28. **使用Java内置函数**:如Arrays.sort()、Collections.sort()等,这些内部优化过的函数通常比自定义实现更快。 29. **使用StringBuilder.append()替换StringBuffer.append()**:在单线程环境中,StringBuilder...
- **复杂性**:相较于 `HashMap`,`ConcurrentHashMap` 的实现更为复杂,因为它需要处理更多并发相关的细节,比如锁机制的实现。 ### 3. 并行与并发的区别 - **并发**:指的是多个任务交替执行的能力,通常由多...
2. **Java实现细节**: - 数据结构:首先,需要定义一个表示数据点的类,包括数据点的坐标(在多维空间中的值)以及所属的簇。同时,还需要一个类来表示簇,存储簇内的数据点和中心。 - 加载数据:从MySQL数据库中...