安装rzsz

yum install -y lrzsz

安装JDK

安装elasticsearch

curl https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.5.2.tar.gz        下载

tar -zxvf elasticsearch-1.5.2.tar.gz        解压

su es    切换启动用户，es不能用root

$ bin/elasticsearch        启动

curl 'http://localhost:9200/?pretty'        测试

curl -XPOST 'http://localhost:9200/_shutdown'        通过API关闭

安装head插件

./plugin --install mobz/elasticsearch-head  #某些版本为install

安装logstash

curl http://download.elastic.co/logstash/logstash/logstash-1.5.0.tar.gz        下载

tar -zxvf logstash-1.5.0.tar.gz

长久以来一直是在linux上配置和使用nginx，竟发现没在本地的windows上装过，今天特意试了一下，把安装和配置记录一下。

首先下载windows版的nginx，我这里是为了学习openresty，所以下载了openresty的windows版， https://openresty.org/download/openresty-1.9.15.1-win32.zip

无需安装，解压到指定目录即可，目录下有nginx.exe，相当于在linux目录中/sbin/nginx的作用。

启动

进入nginx所在目录

1、start nginx
2、nginx.exe
两种方式，第二种或锁定命令行（cmd或者powershell），停服务需要再开启一个终端。建议第一种，这样命令行可以重新接受其他命令。

停止

1、nginx.exe -s stop
2、nginx.exe -s quit
网上资料有说这两种的区别：stop是快速停止nginx，可能并不保存相关信息；quit是完整有序的停止nginx，并保存相关信息。

重载配置

和linux中相同：nginx.exe -s reload

正常的编译安装

wget http://openresty.org/download/ngx_openresty-1.7.4.1.tar.gz
tar -xvf ngx_openresty-1.7.4.1.tar.gz
cd ngx_openresty-1.7.4.1
./configure --prefix=/opt/openresty
make && make install

这里注意，make的时候报错：undefined reference to ‘pcre_free_study’，猜测可能是pcre库找不到的问题，在https://sourceforge.net/projects/pcre/下载pcre，并安装。

安装后，在configure中增加参数–with-pcre=../pcre-8.35，其中注意修改自己pcre库的目录即可。

安装之后，测试下helloworld。例子来自官网。

mkdir ~/work
cd ~/work
mkdir logs/ conf/

新建配置文件nginx.conf

worker_processes  1;
error_log logs/error.log;
events {
    worker_connections 1024;
}
http {
    server {
        listen 8085;
        location / {
            default_type text/html;
            content_by_lua '
                ngx.say("<p>hello, world</p>")
            ';
        }
    }
}

在linux环境变量path中，增加/opt/openresty/nginx/sbin

通过vi /etc/profile, 在文件中追加内容，source /etc/profile。

启动测试的nginx

nginx -p `pwd`/ -c conf/nginx.conf

测试

curl http://localhost:8085/

显示的结果应该是

<p>hello, world</p>

代码说明

content_by_lua    后续lua代码
content_by_lua_file  后续lua代码源文件
lua_code_cache off；     默认是on，关闭后，会取消对lua代码的缓存，实时加载代码的更新，影响性能

常用文档

https://www.nginx.com/resources/wiki/modules/lua/   
https://github.com/openresty/lua-nginx-module         

Lua的执行阶段

1. init_by_lua http
2. set_by_lua server, server if, location, location if
3. rewrite_by_lua http, server, location, location if
4. access_by_lua http, server, location, location if
5. content_by_lua location, location if
6. header_filter_by_lua http, server, location, location if
7. body_filter_by_lua http, server, location, location if
8. log_by_lua http, server, location, location if

$ bin/elasticsearch

这条命令在*nix系统中，会在前台启动elasticsearch。如果需要后台启动，请添加-d参数。

$ bin/elasticsearch -d -p pid 
$ kill `cat pid` 

es可以将自己的PID打印到文件中，-p参数，指定pid存储的文件（pid文件中）

安装

1. 下载：curl -L -O https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.4/elasticsearch-2.3.4.tar.gz
2. 解压：tar -xvf elasticsearch-2.3.4.tar.gz
3. cd elasticsearch-2.3.4/bin
4. 运行：./elasticsearch

在此，运行es必须在非root用户下，否则会报错：不能在root用户下启动elasticsearch，所以通过以下步骤增加es用户，并修改拥有和组策略。

1. adduser es  #添加es用户
2. chown -R es elasticsearch-2.3.4
3. chgrp -R es elasticsearch-2.3.4

可以指定cluster和node的名称：

./elasticsearch --cluster.name my_cluster_name --node.name my_node_name

org.springframework

核心容器

Spring-Core
Spring-Beans
Spring-Context
Spring-Context-Support
Spring-Expression

AOP

Spring-AOP
Spring-Aspects

Messaging

Spring-Messaging

Web

Spring-Web
Spring-Webmvc
Spring-WebSocket
Spring-Webmvc-Portlet

Data Access / Integration

Spring-JDBC
Spring-TX
Spring-ORM
Spring-OXM
Spring-JMS

https://www.elastic.co/downloads/kibana

首先安转配置启动kibana

1. wget https://artifacts.elastic.co/downloads/kibana/kibana-5.0.0-linux-x86_64.tar.gz
2. 修改config/kibana.yml，elasticsearch.url修改为es的地址，server.host修改为kibana对我暴露的访问地址
3. bin/kibana
4. 通过浏览器 http://host:5601 访问

接下来将讨论以下问题：

1. 加载es中的样本数据
2. 定义index pattern
3. 通过discover展示数据
4. 为样本数据启动可视化
5. 将可视化图表组装成dashboard

execute

线程池是通过execute方法，添加任务（task）的。方法保证在将来的某一时间，执行给定的任务。任务可能在新的线程中被执行，也可能在已经存在于池子中的线程中被执行。

如果任务因为线程池关闭，或者容量满，而不能被提交执行，任务将被当前设置的RejectedExecutionHandler处理。

后续的执行步骤是三种情况

1. 如果运行的线程数小于corePoolSize，则尝试新建一个线程，并设置当前的任务为该线程第一个需要处理的任务。
2. 如果运行的线程数大于等于corePoolSize，则将该任务添加到阻塞队列workQueue中，池子中的空闲线程就会去队列汇总取任务之心。
3. 当队列满了后，添加队列会失败，这是尝试直接创建一个线程，如果失败，可以得知，要么线程池已经关闭，要么是达到了饱和状态，不能在接受新任务，则该任务将被拒绝。

execute代码

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    // 上面的step1
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 上面的step2，workQueue.offer可以把command添加到阻塞队列中，这个地方有双检测
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // double-checked，如果线程池关闭，还要remove
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 如果是running状态 c & CAPACITY = e000 0000 & 1000 0000 = 0
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

addWorker

该方法用于创建一个新线程，即上面第一种情况。addWorker(firstTask,true)

addWorker代码

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        // api说明中的解释是，如果线程池关闭或适合关闭，则方法返回false
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 如果worker数量比coresize大，或者比maxsize大，就不需要添加了
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 对clt做cas自增，成功后跳出外层循环retry；失败的话，说明有其他线程修改了clt，继续后续代码逻辑    
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl 重新获取ctl
            if (runStateOf(c) != rs) // 不同，说明ctl变化，continue外层retry循环
                continue retry;
            // 否则，说明cas更新失败，应该重新内存循环，尝试再一次cas自增
        }
    }

    // works是个存有work对象的hashset，为了保证线程安全，需要通过mainlock加锁，保证同步访问,当调用方，传入形参addWorker(null,false)的时候，是不会触发后面的加锁同步逻辑的
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int c = ctl.get();
                int rs = runStateOf(c);

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

任务通过execute方法加入到线程池后，调用链是：

execute –> addWorker –> t.start –> 内部类worker.run() –> runWorker –> getTask() –> workQueue.take 到这里就可以完成从队列中取任务执行，并且可以阻塞。具体何时getTask由上层方法调用决定，即step1，2，3。

同步框架AQS

作为同步基础框架，需要提供：维护一个同步状态（表示锁被锁定或释放），包含对状态的更新和检查操作，包含能够阻塞当前线程的方法，其他线程修改同步状态后能恢复。

在AQS中Lock和Condition的组合，约等于Synchronized和wait/notify的组合。

conditiong的作用是维护等待信号队列，并在适当的时机加入到AQS等待队列中以实现唤醒。

AQS需满足的三要素

• 自动维护同步状态
• 阻塞和唤醒线程
• 维护队列

Synchronization State，是32位的int，使用volatile保证state在线程间可见，并通过CAS操作完成更新，保证变量的同步。compareAndSetState与期望值对比，相同才更新。

Blocking，提供独占（ReentrantLock）和共享模式（Semaphore），以及同时包含两种模式的ReentrantReadWriteLock。

Queue，框架的核心就是维护的这个包含阻塞线程的队列。队列的结构可以参考：

Doug Lea, The java.util.concurrent Synchronizer Framework, Science of Computer Programming, 2005, 58(3):293-309

AQS应用

一般是声明私有内部类去继承AQS，并代理AQS的全部或者部分方法。后面将会看到ReentrantLock，闭锁、信号量、屏障、读写锁，都是这样实现的。

Java中需要序列化的类，只需实现Serializable接口，具体的实现由虚拟机负责调用相关实现方式去实现。所有的non-transient和non-static都将被序列化。

class SessionDTO implements Serializable {
    private static final long serialVersionUID = 1L;
    private int data; // Stores session data

    // Session activation time (creation, deserialization)
    private long activationTime; 

    public SessionDTO(int data) {
        this.data = data;
        this.activationTime = System.currentTimeMillis();
    }

    public int getData() {
        return data;
    }

    public long getActivationTime() {
        return activationTime;
    }
}

public class SerializeTester implements Serializable {
    public static void main(String... strings) throws Exception {
        File file = new File("out.ser");

        ObjectOutputStream oos = new ObjectOutputStream(
            new FileOutputStream(file));
        SessionDTO dto = new SessionDTO(1);
        oos.writeObject(dto);
        oos.close();

        ObjectInputStream ois = new ObjectInputStream(
            new FileInputStream(file));
        SessionDTO dto = (SessionDTO) ois.readObject();

        System.out.println("data : " + dto.getData()
            + " activation time : " + dto.getActivationTime());
        ois.close();
    }
}

某些情况下，如果某些字段不需要进行序列化，可以将其设置为transient，这样这些字段就不会进行序列化，以减轻对网络带宽的占用。某些值，如果需要在反序列化的时候重新赋值，可以设置为transient，并且重写readObject方法，在其中为该属性赋值。如果希望序列化的时候对某些值做自定义的处理，可以重写writeObject方法，添加相应逻辑。

class SessionDTO implements Serializable {
    private static final long serialVersionUID = 1L;
    private transient int data; // Stores session data

    //Session activation time (creation, deserialization)
    private transient long activationTime; 

    public SessionDTO(int data) {
        this.data = data;
        this.activationTime = System.currentTimeMillis();
    }

    private void writeObject(ObjectOutputStream oos) throws IOException {
        oos.defaultWriteObject();
        oos.writeInt(data);
        System.out.println("session serialized");
    }

    private void readObject(ObjectInputStream ois) throws IOException,
            ClassNotFoundException {
        ois.defaultReadObject();
        data = ois.readInt();
        activationTime = System.currentTimeMillis();
        System.out.println("session deserialized");
    }

    public int getData() {
        return data;
    }

    public long getActivationTime() {
        return activationTime;
    }
}

两个方法都是private的，并且它们既不存在于java.lang.Object，也没有在Serializable中声明。那么ObjectOutputStream如何使用它们的呢？这个吗，ObjectOutputStream使用了反射来寻找是否声明了这两个方法。因为ObjectOutputStream使用getPrivateMethod，所以这些方法不得不被声明为priate以至于供ObjectOutputStream来使用。