Java 第3篇

集合

Java提供了丰富的集合库。

 

集合是一种常见的数据结构。在计算机领域，集合是一组可变数量的数据项的组合。当我们定义一个集合时，集合中的数据项通常都具有某些共同的特征。正是它们具有某些共同的特征，才是我们对它们定义一个集合并对它们进行研究。集合在很多领域都有研究，在数学领域，集合论就是对集合研究的一个分支。

 

数学上，集合定义为

S = {x1,x2,...}

也可以定义为

S = {x : some condition on x is met}

将集合描述为满足某种规则的对象集S，x表示满足这个集合的元素。

 

集合可以是空集，定义为

φ = {}

 

写道

A set is an unordered collection of things, its elements. Write x ∈ S or S ∋ x for an element x of S.

 

写道

[TEC HAL] NAIVE SET THEORY
https://kyl.neocities.org/books/[TEC%20HAL]%20naive%20set%20theory.pdf

 

写道

SetTheoryDover- Charles C Pinter
http://matematicas.uis.edu.co/adrialba/sites/default/files/SetTheoryDover-%20Charles%20C%20Pinter.pdf

集合，专业术语称之为Set。数学上，有些资料上也有叫Collection或者Class。

 

Collection

Java中的集合统称为Collection。提供了丰富的实现，通常，狭义上的集合一般指的是上面讲的集合，即我们常说的Set。Java中的集合有更广泛的意义。在Java中，List、Queue、Deque等都认为是一种集合。

 

Java提供的所有集合都有一个最基本的接口，即Collection。

Collection

public interface Collection<E> {
    ...
}

Collection接口扩展了Iterable。这是一种遍历器模式的体现，当我们需要遍历集合中的元素时，一个要考虑的因素就是不用考虑集合结构的内部实现，甚至在遍历的时候不会破坏集合的内部结构，而可以方便的遍历集合中的元素。

public interface Collection<E> extends Iterable<E> {
    ...
}

Iterator<E> iterator();

Spliterator<E> spliterator();

在Iterable接口中还定义了一个forEach方法

    default void forEach(Consumer<? super T> action) {
        Objects.requireNonNull(action);
        for (T t : this) {
            action.accept(t);
        }
    }

 

 

Object类中有两个非常重要的方法，在Collection接口中也定义了这两个方法。这两个方法的定义和Object类中一致。在Collection接口中定义的本意是当通过一个集合接口引用，比如Collection接口引用，对集合进行操作的时候，可以对集合进行比较。

 

boolean equals(Object o);

int hashCode();

 

List

ArrayList

Vector

 

链表

 

头节点

 

哨兵节点

 

Set

HashSet

 

Map

HashMap

Hashtable

 

队列

队列是一种在队尾部插入，在队头部删除的先入先出（FIFO）数据结构。在队尾部插入称为入队，在队头部删除称为出队。

 

队列最重要的两个操作就是入队和出队操作。

 

Java定义了一个Queue接口，这是一个队列接口，Java提供的所有队列都实现了这个接口。这个接口也扩展了Collection接口，Java认为队列也是一种集合。

 

队列是一个典型的生产者消费者问题。生产者向队列生产数据，消费者从队列消费数据。如果队列满了，生产者必须进入等待，或者进入睡眠，先睡一下，等待消费者消费数据以空出队列空间。如果队列空了，消费者也必须进入等待，或者也进入睡眠，等到生产者向队列生产数据才能继续消费。

 

AbstractQueue

AbstractQueue抽象类实现了Queue接口，并继承AbstractCollection抽象类。

 

阻塞队列

Java定义了一个BlockingQueue接口，表示一个阻塞队列。这个接口扩展了Queue接口。

 

ArrayBlockingQueue

 

LinkedBlockingQueue

LinkedBlockingQueue内部结构采用的是单链表。链表结构由一组不连续的节点组成，每个节点都有一个指向后续节点（successor）的指针或引用。这个链表结构由LinkedBlockingQueue内部的一个静态类Node实现。

    static class Node<E> {
        E item;

        /**
         * One of:
         * - the real successor Node
         * - this Node, meaning the successor is head.next
         * - null, meaning there is no successor (this is the last node)
         */
        Node<E> next;

        Node(E x) { item = x; }
    }

LinkedBlockingQueue在构造时，就是构造初始化一个链表。这个链表在构造时为了方便分配了一个头节点，这个头节点起到哨兵的角色，称为哨兵节点。这个节点不存储队列数据。

    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

 

在入队时就是在链表尾部插入一个节点，出队就是将链表头部节点删除。在入队往队列中插入数据时，需要先对队列加锁锁定，以防止有其他线程也往队列中插入数据。

 

LinkedBlockingQueue的入队操作提供了多个方法，调用put、offer方法都可以在队尾生产数据。

 

put

public void put(E e) throws InterruptedException

这个方法在队尾插入数据，如果队列满了，将进入等待直到消费者线程消费数据以空出队列空间。期间如果被其他线程中断，则报中断异常返回。这个方法没有返回值。

 

在往队列中插入数据时，put调用ReentrantLock的lockInterruptibly方法先加锁锁定，这个方法可能会报中断异常InterruptedException，在锁定期间，如果有线程中断生产线程，通过调用生产线程的interrupt方法，lockInterruptibly方法报中断异常。

 

如果队列满了，调用ConditionObject的await方法进入等待，ConditionObject类实现了Condition接口。await方法也可能会报中断异常InterruptedException。如果生产者线程处于等待状态，当消费者线程消费数据以空出了多余的队列空间时，消费者线程唤醒生产者线程，这时候生产者线程继续执行，向队列中生产插入数据。在生产者等待期间，如果有线程中断生产线程，通过调用生产线程的interrupt方法，await方法报中断异常返回。

 

offer

这个方法有两种实现，其中一个指定了一个等待超时时间。

public boolean offer(E e)

这个方法没有指定等待超时时间。

 

public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException

这个方法指定了一个等待超时时间。

 

LinkedBlockingQueue的出队操作也提供了多个方法，调用take、poll方法都可以从队头消费数据。另外还有一个peek方法，这个方法在消费数据的时候，不会将数据从队头删除。

 

take

public E take() throws InterruptedException

 

poll

这个方法也有两种实现，其中一个指定了一个等待超时时间。

 

public E poll()

 

public E poll(long timeout, TimeUnit unit) throws InterruptedException

 

peek

public E peek()

这个方法只消费但不会将数据从队列中删除。

 

非阻塞队列

ConcurrentLinkedQueue

 

ConcurrentLinkedDeque

 

双向队列

 

Stack

Stack也就是我们说的栈，是一种先进后出的结构。最重要的两个操作就是push和pop，即入栈和出栈操作。栈有一个栈底和栈顶，栈底指向栈底的位置，栈顶指向栈顶的位置。初始时，栈顶和栈底指向相同的位置，当有数据被压入栈中，栈顶向栈顶移动一个位置，出栈时栈顶向栈底移动一个位置。

 

Java中Stack继承至Vector。Stack采用的是顺序表实现，采用这种方式的栈空间是一块连续的存储空间。我们常说的函数栈也是这种结构。入栈和出栈只是简单的将栈顶向上或向下加一减一。

 

push

public E push(E item)

 

pop

public synchronized E pop()

 

peek

public synchronized E peek()

这个方法不会将数据从栈中弹出。

 

多线程

 

线程

Java线程包括用户线程和守护线程。

 

通常我们创建的线程都是用户线程。创建的线程是用户线程还是守护线程由daemon指定，两种情况决定创建的线程是用户线程还是守护线程，如果创建线程时的父线程是守护线程，那么创建的线程也是守护线程。可以在创建线程时通过指定daemon来创建一个守护线程。

 

守护线程

Java守护线程是伴随着进程的。Java进程运行结束，其中的守护进程也会跟着结束，它跟用户线程不同，Java进程运行要结束，需要等待其中的用户线程结束才能结束。

public static void main(String[] args) {
    Thread t = new Thread() {
        public void run() {
            while (true) {
                // 
            }
        }
    };
    t.setDaemon(true);
    t.start();
}

 

 

线程状态

 

线程状态转换

 

线程中断状态

我们在上面讲解了线程状态，这里再讲解一个线程相关的状态：线程中断状态（thread's interrupt status）。

 

线程中断状态相关的操作interrupt方法，它和计算机中的中断是不一样的。计算机中的中断会中断当前程序的执行，会让出CPU，以便执行其他程序。

 

每一个线程都有一个对应的线程中断状态（thread's interrupt status）

 

线程返回值

在Java中，我们通过Thread创建线程，线程启动后执行的是run方法，该方法实现如下：

    public void run() {
	if (target != null) {
	    target.run();
	}
    }

该方法定义在Runnable接口中，Thread实现了这个接口。从该方法定义可以看出，它是没有返回值的。这跟Posix的线程规范不一样，我们知道在linux中，创建线程时指定的线程函数是有返回值的，线程终止时是可以得到这个返回值的。

 

针对线程返回值的情况，可参考FutureTask，以下是一个简单的例子：

 

public class FutureTaskTest1 {

    public static void main(String[] args) throws Exception {
        Callable<String> task = new Callable<String>() {
            public String call() throws Exception {
                System.out.println("do something");
                return "something";
            }
        };
        FutureTask<String> op = new FutureTask<String>(task);
        Thread th = new Thread(op);
        th.start();
        if (! op.isDone()) {
//            if (! op.cancel(true)) {
//                if (op.isDone()) {
//                    // done
//                } else if (op.isCancelled()) {
//                    // already cancelled
//                } else {
//                    throw new Exception("fail to cancel.");
//                }
//            }
//            throw new Exception("cancelled.");
        }
        String result = op.get();
        System.out.println(result);
    }
}

 

 

 

线程操作

 

中断

 

    public void interrupt() {
	if (this != Thread.currentThread())
	    checkAccess();

	synchronized (blockerLock) {
	    Interruptible b = blocker;
	    if (b != null) {
		interrupt0();		// Just to set the interrupt flag
		b.interrupt();
		return;
	    }
	}
	interrupt0();
    }

 

 

 

private native void interrupt0();

 

yield

 

join

join方法将使得当前线程Blocked，当前线程处于WAITING状态，确切的说线程处于等待队列当中，因为join内部其实也是调用的wait方法，等待直到对应线程结束消亡，当前线程才能继续往下执行。如t.join();直到线程t结束消亡，当前线程才能继续往下执行。

 

除了等待直到对应线程结束消亡，当前线程才能继续往下执行，如果有其他线程通过interrupt方法中断当前线程，将触发中断异常，也能使当前线程结束等待继续执行。

 

public final void join() throws InterruptedException {
    join(0);
}

 

 

public final synchronized void join(long millis)
throws InterruptedException {
    long base = System.currentTimeMillis();
    long now = 0;

    if (millis < 0) {
        throw new IllegalArgumentException("timeout value is negative");
    }

    if (millis == 0) {
        while (isAlive()) {
            wait(0);
        }
    } else {
        while (isAlive()) {
            long delay = millis - now;
            if (delay <= 0) {
                break;
            }
            wait(delay);
            now = System.currentTimeMillis() - base;
        }
    }
}

join例子：

final Thread main = Thread.currentThread();
Thread t = new Thread() {
    public void run() {
        long st = System.currentTimeMillis();
        while (true) {
            if (System.currentTimeMillis() - st > 5 * 60 * 1000) {
                main.interrupt();
                break;
            }
        }
    }
};
t.start();
try {
    t.join();
} catch (InterruptedException e) {
    e.printStackTrace();
}

运行结果

>java JoinTest
java.lang.InterruptedException
        at java.lang.Object.wait(Native Method)
        at java.lang.Thread.join(Thread.java:1245)
        at java.lang.Thread.join(Thread.java:1319)
        at JoinTest.main(JoinTest.java:27)

在程序退出之前jstack dump结果：

"main" #1 prio=5 os_prio=0 tid=0x0088a800 nid=0x137c in Object.wait() [0x009bf00

0]

   java.lang.Thread.State: WAITING (on object monitor)

        at java.lang.Object.wait(Native Method)

        - waiting on <0x02cd8918> (a JoinTest$1)

        at java.lang.Thread.join(Thread.java:1245)

        - locked <0x02cd8918> (a JoinTest$1)

        at java.lang.Thread.join(Thread.java:1319)

        at JoinTest.main(JoinTest.java:27)

 

这里WAITING (on object monitor)，这个waiting的对象就是下面：waiting on <0x02cd8918>，这里的0x02cd8918实际上就是线程对象t的内存地址。可以验证下，我们在上面的代码上打印出对象t的内存地址：

final Thread main = Thread.currentThread();
Thread t = new Thread() {
    public void run() {
        long st = System.currentTimeMillis();
        while (true) {
            if (System.currentTimeMillis() - st > 5 * 60 * 1000) {
                main.interrupt();
                break;
            }
        }
    }
};
t.start();
long baseAddress = location(t);
System.out.println("\"" + t.getName() + "\":0x" + Long.toHexString(baseAddress));
try {
    t.join();
} catch (InterruptedException e) {
    e.printStackTrace();
}

其中long baseAddress = location(t);获取的是t的内存地址，参考另一篇文章：https://lobin.iteye.com/blog/2327679。

运行输出

>java JoinTest
"Thread-0":0x2cd8918
java.lang.InterruptedException
        at java.lang.Object.wait(Native Method)
        at java.lang.Thread.join(Thread.java:1245)
        at java.lang.Thread.join(Thread.java:1319)
        at JoinTest.main(JoinTest.java:27)

输出的0x2cd8918正好和waiting on <0x02cd8918>是一样的。

 

线程安全

关于线程安全，我的理解和判定标准是：

写道

一个函数或对象不管怎么使用都能保证其正确性。

 

有关线程安全可参考另一篇文章：https://lobin.iteye.com/blog/2322896。

 

 

写道

这个函数是线程安全的吗?

 

 

 

 

写道

这个方法是线程安全的吗?

 

 

 

 

写道

这个对象是线程安全的吗?

 

有关线程安全可参考另一篇文章：https://lobin.iteye.com/blog/2322896.

 

和线程安全相关的另一个概念：可重入。

写道

Reentrancy is distinct from, but closely related to, thread-safety. A function can be thread-safe and still not reentrant. For example, a function could be wrapped all around with a mutex (which avoids problems in multithreading environments), but if that function is used in an interrupt service routine, it could starve waiting for the first execution to release the mutex. The key for avoiding confusion is that reentrant refers to only one thread executing. It is a concept from the time when no multitasking operating systems existed.

 

 

可重入

 

 

线程池

Java中的线程池实现类ThreadPoolExecutor。

ThreadPoolExecutor线程池通过启动一组工作线程Worker来执行线程池中的线程，工作线程并不是在初始化线程池时就已经启动好的，在初始化线程池时，需要指定的参数:

corePoolSize

maximumPoolSize

keepAliveTime

unit

workQueue

threadFactory

handler

其中corePoolSize表示

 

 

 

线程池运行状态

1、RUNNING

2、SHUTDOWN

3、STOP

4、TERMINATED
在java 1.6中通过一个volatile变量runState来表示线程池当前的运行状态。在java 1.8中运行状态封装在控制状态ctl中，它是一个原子变量，控制状态ctl封装了运行状态runState和工作线程个数workerCount。

java 1.8中新增了一个状态

5、TIDYING

 

工作线程

    private final class Worker implements Runnable {
        /**
         * The runLock is acquired and released surrounding each task
         * execution. It mainly protects against interrupts that are
         * intended to cancel the worker thread from instead
         * interrupting the task being run.
         */
        private final ReentrantLock runLock = new ReentrantLock();

        /**
         * Initial task to run before entering run loop. Possibly null.
         */
        private Runnable firstTask;

        /**
         * Per thread completed task counter; accumulated
         * into completedTaskCount upon termination.
         */
        volatile long completedTasks;

        /**
         * Thread this worker is running in.  Acts as a final field,
         * but cannot be set until thread is created.
         */
        Thread thread;

        Worker(Runnable firstTask) {
            this.firstTask = firstTask;
        }

        boolean isActive() {
            return runLock.isLocked();
        }

        /**
         * Interrupts thread if not running a task.
         */
        void interruptIfIdle() {
            final ReentrantLock runLock = this.runLock;
            if (runLock.tryLock()) {
                try {
		    if (thread != Thread.currentThread())
			thread.interrupt();
                } finally {
                    runLock.unlock();
                }
            }
        }

        /**
         * Interrupts thread even if running a task.
         */
        void interruptNow() {
            thread.interrupt();
        }

        /**
         * Runs a single task between before/after methods.
         */
        private void runTask(Runnable task) {
            final ReentrantLock runLock = this.runLock;
            runLock.lock();
            try {
                /*
                 * Ensure that unless pool is stopping, this thread
                 * does not have its interrupt set. This requires a
                 * double-check of state in case the interrupt was
                 * cleared concurrently with a shutdownNow -- if so,
                 * the interrupt is re-enabled.
                 */
                if (runState < STOP &&
                    Thread.interrupted() &&
                    runState >= STOP)
                    thread.interrupt();
                /*
                 * Track execution state to ensure that afterExecute
                 * is called only if task completed or threw
                 * exception. Otherwise, the caught runtime exception
                 * will have been thrown by afterExecute itself, in
                 * which case we don't want to call it again.
                 */
                boolean ran = false;
                beforeExecute(thread, task);
                try {
                    task.run();
                    ran = true;
                    afterExecute(task, null);
                    ++completedTasks;
                } catch (RuntimeException ex) {
                    if (!ran)
                        afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                runLock.unlock();
            }
        }

        /**
         * Main run loop
         */
        public void run() {
            try {
                Runnable task = firstTask;
                firstTask = null;
                while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);
            }
        }
    }

 

 

 

 

其他相关实现还有ScheduledThreadPoolExecutor。

 

ScheduledThreadPoolExecutor实现用于对要执行的线程任务延迟执行、周期性执行。

 

 

 

线程模型

Java内存模型

 

用户线程

 

轻量级线程

 

内核线程

 

Java线程

 

线程池

 

 

锁

锁是一种非常普遍，并且经常使用的一种技术。在一个多任务操作系统中，如果多个进程同时访问资源，就会产生竞争。

 

在多线程程序中，多个线程同时访问某个共享数据时，就会出现上面提到的竞争问题。

 

为了解决这个竞争问题，需要保证竞争访问共享资源的多个进程能够同步有序的对共享资源进行操作。

 

锁是一种解决多个进程竞争访问共享资源的一种同步机制。

 

 

 

ReentrantLock

可重入锁

ReentrantLock中的Reentrant怎么理解？

https://www.geeksforgeeks.org/reentrant-function/ 写道

A function is said to be reentrant if there is a provision to interrupt the function in the course of execution, service the interrupt service routine and then resume the earlier going on function, without hampering its earlier course of action. Reentrant functions are used in applications like hardware interrupt handling, recursion, etc.

 

 

可指定公平锁（fair lock）和非公平锁（non-fair lock）。其内部是通过Sync来实现的，它有两个实现：FairSync和NonfairSync，分别实现公平锁（fair lock）和非公平锁（non-fair lock）。

 

ReentrantLock通过AbstractQueuedSynchronizer（Sync继承了AbstractQueuedSynchronizer）中的state表示同步状态。

 

ReentrantLock还通过一个等待队列（wait queue）来保存想要获取锁，但锁已被其他线程占用，暂时还获取不到锁的线程。等待队列使用了一种叫做"CLH" (Craig, Landin, and Hagersten)锁队列（lock queue）。

java.util.concurrent.locks.AbstractQueuedSynchronizer.Node 写道

The wait queue is a variant of a "CLH" (Craig, Landin, and Hagersten) lock queue. CLH locks are normally used for spinlocks. We instead use them for blocking synchronizers, but use the same basic tactic of holding some of the control information about a thread in the predecessor of its node.

 

 

等待队列（wait queue）通过AbstractQueuedSynchronizer.Node类实现。

 

 

 

例子：

public class ReentrantLockTest1 {

    private static Lock lock;

    private static void dosomething() {
        lock.lock();
        lock.lock();

        System.out.println("do something.");

        lock.unlock();
    }

    public static void main(String[] args) throws Exception {
        lock = new ReentrantLock();

        dosomething();
        dosomething();

        System.out.println("end.");
    }
}

例子：

public class ReentrantLockTest2 {

    private static Lock lock;

    private static void dosomething(Thread main) {
        lock.lock();
        lock.lock();

        System.out.println(Thread.currentThread().getName() + ": do something.");

        lock.unlock();
        if (main != null) {
            main.interrupt();
        }
    }

    public static void main(String[] args) throws Exception {
        lock = new ReentrantLock();
        Thread main = Thread.currentThread();
        System.out.println("main thread: " + main.getName());

       Thread t1 = new Thread() {
           public void run() {
               dosomething(main);
           }
       };
       t1.start();

        try {
            t1.join();
        } catch (InterruptedException e) {
            System.out.println(Thread.currentThread().getName() + ": interrupted.");
        }
        dosomething(null);

        System.out.println("end.");
    }
}

 

公平锁（fair lock）

 

非公平锁（non-fair lock）

 

 

同步

关于同步，可参考我另一篇文章：https://lobin.iteye.com/blog/1558775

 

AQS

java.util.concurrent.locks.AbstractQueuedSynchronizer

 

java synchronizer

 

java.util.concurrent.locks.AbstractOwnableSynchronizer

 

java.util.concurrent.locks.AbstractQueuedSynchronizer

 

java.util.concurrent.CountDownLatch.Sync

 

java.util.concurrent.FutureTask.Sync

 

java.util.concurrent.locks.ReentrantLock.Sync

 

java.util.concurrent.locks.ReentrantLock.FairSync

 

java.util.concurrent.locks.ReentrantLock.NonfairSync

 

java.util.concurrent.locks.ReentrantReadWriteLock.Sync

 

java.util.concurrent.locks.ReentrantReadWriteLock.FairSync

 

java.util.concurrent.locks.ReentrantReadWriteLock.NonfairSync

 

java.util.concurrent.Semaphore.Sync

 

java.util.concurrent.Semaphore.FairSync

 

java.util.concurrent.Semaphore.NonfairSync

 

 

java.util.concurrent.locks.AbstractQueuedLongSynchronizer

 

 

阻塞同步（互斥同步）

 

synchronized

synchronized可以作为修饰符作用在方法上，它还是一个同步原语在方法内部使用。

synchronized也是可重入的，参考ReentrantLock的可重入。

 

作为修饰符作用在方法上：

public synchronized void doanything() {
	System.out.println("do doanything.");
}

 

在方法内部使用：

public void dosomething() {
	synchronized (this) {
		System.out.println("do something.");
	}
}

 

 

@Test
public void testSynchronized() {
	synchronized (this) {

		synchronized (this) {
			System.out.println("do something.");
		}

	}
}

 

非阻塞同步

 

自旋锁

 

CAS

参考另一篇文章：https://lobin.iteye.com/blog/2325737

 

阻塞

阻塞与非阻塞其实是和进程(线程)运行状态有关的概念是。一般情况下，除了正常的中断情况发生，进程(线程)是无阻塞的运行，也就是非阻塞。如果进程(线程)运行到某个时刻因需要等待每个资源或条件发生而无法继续运行，这就是阻塞。

 

无锁编程

 

并发编程

 

并发包

Java SDK为并发场景提供的编程库（api），也就是我们常用到的并发包（java.util.concurrent）。

 

线程池

原子变量

同步

锁

条件变量

信号量

屏障

public class CyclicBarrierTest extends Thread {

    public void run() {
        System.out.println("do something.");
    }

    public static void main(String[] args) throws Exception {
        CyclicBarrierTest action = new CyclicBarrierTest();
        CyclicBarrier cb = new CyclicBarrier(1, action);
//        CyclicBarrier cb = new CyclicBarrier(5, action);
        cb.await();
//        cb.await();
//        cb.await();
//        cb.await();
//        cb.await();
    }
}

 

CopyOnWriteArrayList

 

CopyOnWriteArraySet

其内部使用的就是CopyOnWriteArrayList。

 

copy-on-write

写时拷贝

 

队列

 

阻塞队列

 

非阻塞队列

ConcurrentLinkedQueue

 

双向队列

 

Stack

 

 

IO

支持面向字节流的i/o输入输出和面向字符流的i/o输入输出。

Java I/O为这两种i/o输入输出提供了统一的接口，面向字节流的i/o输入输出接口：InputStream，OutputStream。面向字符流的i/o输入输出接口：Reader，Writer。

 

面向字节流的i/o输入输出

 

面向字符流的i/o输入输出

 

NIO

1、HeapByteBuffer

2、DirectByteBuffer

 

DirectByteBuffer和HeapByteBuffer之间的本质区别时DirectByteBuffer是在直接内存（堆外内存）分配，HeapByteBuffer是在堆（Heap）上分配。

在Java中，堆外内存是通过Unsafe的allocateMemory来分配内存的，使用完后，它需要手动进行释放。通过DirectByteBuffer在直接内存（堆外内存）分配的内存空间是不需要手动释放的。

 

Select

 

序列化

-><magic,2 bytes,><version,2 bytes,>

<magic> -> 0xACED

<version> -> 0x0005 -> 5

 

1、NULL

2、字符串

3、数组

4、枚举

5、字节数据

6、Class

7、对象

-><magic,2 bytes,><version,2 bytes,><,1 byte,0x73><,1 byte,0x72><类描述信息,n bytes,><,1 byte,0x78><父类描述信息,n bytes,><基础类型字段值,n bytes,有多少个字段就有多少个><非基础类型字段值,n bytes,有多少个字段就有多少个>

<magic> -> 0xACED

<version> -> 0x0005 -> 5

<类描述信息,n bytes,> -> <类名,n bytes,><Serial Version,8 bytes,><标志位,1 byte,><字段数,n bytes,><字段信息,n bytes,有多少个字段就有多少个。>

<父类描述信息,n bytes,> 同 <类描述信息,n bytes,>

 

<类名,n bytes,> -> <类名(字符串)长度,2 byte,类名(字符串)长度为类名(字符串)的UTF编码之后的字节数><类名(字符串)信息,n bytes,为类名(字符串)的UTF编码>

 

<字段信息,n bytes,有多少个字段就有多少个。> -> <字段类型标识（signature）,1 byte,><字段名称(字符串),n bytes,><字段类型信息（signature）,n byte,非基础类型才有这个字段>

 

<字段名称(字符串),n bytes,> -> <字段名称(字符串)长度,2 byte,字段名称(字符串)长度为字段名称(字符串)的UTF编码之后的字节数><字段名称(字符串)信息,n bytes,为字段名称(字符串)的UTF编码>

<字段类型信息（signature）,n byte,非基础类型才有这个字段> -> <字段类型信息(字符串)长度,2 byte,字段类型信息(字符串)长度为字段类型信息(字符串)的UTF编码之后的字节数><字段类型信息(字符串)信息,n bytes,为字段类型信息(字符串)的UTF编码>

 

字段类型信息（signature）和字段类型标识（signature）区别：

如：

Ljava/util/Date;

字段类型标识（signature）为L，字段类型信息（signature）为Ljava/util/Date;

Ljava/lang/String;

字段类型标识（signature）为L，字段类型信息（signature）为Ljava/lang/String;

例子：

public class TestObject implements Serializable {

    private int id;

    private String name;

    private float price;

    private double marketPrice;

    private Date creation;
}

 

 

public static void main(String[] args) throws IOException {
    TestObject data = new TestObject();
    writeToFile(data, "data");
}

private static void writeToFile(TestObject data, String filename) throws IOException {
    ObjectOutputStream os = new ObjectOutputStream(new FileOutputStream(filename));
    os.writeObject(data);
}

 

 

 

 

 

CountDownLatch

 

CountDownLatch是Java并发包中用于并发编程的同步辅助类。用于允许一个或多个线程等待其他线程中一组操作执行完成，在其他线程中的这组操作执行完成之前，等待的线程将阻塞无法继续执行；或者等待的时间到了才能继续执行；或者等待的线程被中断。

 

CountDownLatch构造：

public CountDownLatch(int count)

在构造CountDownLatch需要传入一个count down的次数。CountDownLatch将使用java.util.concurrent.CountDownLatch.Sync进行同步控制。java.util.concurrent.CountDownLatch.Sync使用AQS状态来表示这个count down。AQS状态可参考java.util.concurrent.locks.AbstractQueuedSynchronizer

 

 

countDown方法：

 

public void countDown()

调用该方法将将count down减1，直到为0后，阻塞在await方法的地方就能继续执行。

 

await方法：

public void await() throws InterruptedException

调用该方法后，当前线程将被阻塞，直到count down数被减到0后或者当前线程被中断才能继续执行。

 

public boolean await(long timeout, TimeUnit unit) throws InterruptedException

调用该方法后，当前线程将被阻塞，直到count down数被减到0后或者当前线程被中断或者等待的时间到了才能继续执行。

 

 

CountDownLatch 实例

package com.java;

 

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.CountDownLatch;

 

abstract class AbstractSql {

public abstract void execute();

}

 

class Sql extends AbstractSql {

private String sql;

 

public Sql(String sql) {

this.sql = sql;

}

 

public String getSql() {

return sql;

}

 

public void execute() {

try {

System.out.println(sql + " is executing...");

Thread.sleep(5000);

System.out.println(sql + " executed!");

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

 

class Result {

 

private Sql sql;

 

public Result(Sql sql) {

this.sql = sql;

}

 

public String get() {

return sql.getSql() + " 's result!";

}

}

 

class Aggregation extends AbstractSql {

 

private List<Sql> sqllist;

 

private CountDownLatch latch;

 

private List<Result> results;

 

public Aggregation(List<Sql> sqllist) {

if (sqllist == null || sqllist.size() < 1) {

throw new IllegalArgumentException("no sql");

}

this.sqllist = sqllist;

latch = new CountDownLatch(sqllist.size());

results = new ArrayList<Result>();

}

 

private class WaitForResult extends Thread {

private Aggregation aggregator;

private Sql sql;

 

public WaitForResult(Sql sql, Aggregation aggregator) {

this.sql = sql;

this.aggregator = aggregator;

}

 

public void run() {

sql.execute();

aggregator.results.add(new Result(sql));

latch.countDown();

}

}

 

private List<Result> forResult() {

for (final Sql sql : sqllist) {

new WaitForResult(sql, this).start();

}

return results;

}

 

private Thread waitForResult() throws InterruptedException {

Thread t = new Thread() {

public void run() {

try {

latch.await();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

};

t.start();

return t;

}

 

private void aggregate() throws InterruptedException {

Thread t = waitForResult();

List<Result> results = forResult();

t.join();

for (Result result : results) {

System.out.println(result.get());

}

System.out.println("all results aggregated!!!");

}

 

@Override

public void execute() {

try {

aggregate();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

 

}

 

public class CountDownLatchTest {

 

public static void main(String[] args) throws Exception {

List<Sql> sqllist = new ArrayList<Sql>();

sqllist.add(new Sql("select 1 from 1"));

sqllist.add(new Sql("select 2 from 2"));

sqllist.add(new Sql("select 3 from 3"));

sqllist.add(new Sql("select 4 from 4"));

sqllist.add(new Sql("select 5 from 5"));

Aggregation agg = new Aggregation(sqllist);

agg.execute();

}

}

 

 

 

运行结果：

select 1 from 1 is executing...

select 3 from 3 is executing...

select 5 from 5 is executing...

select 4 from 4 is executing...

select 2 from 2 is executing...

select 1 from 1 executed!

select 3 from 3 executed!

select 5 from 5 executed!

select 4 from 4 executed!

select 2 from 2 executed!

select 1 from 1 's result!

select 3 from 3 's result!

select 5 from 5 's result!

select 4 from 4 's result!

select 2 from 2 's result!

all results aggregated!!!