Nicksxs's Blog

What hurts more, the pain of hard work or the pain of regret?

0%

Posted on Edited on In Views: Disqus:

最近看了大神的 AQS 的文章，之前总是断断续续地看一点，每次都知难而退，下次看又从头开始，昨天总算硬着头皮看完了第一部分
首先 AQS 只要有这些属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 头结点，你直接把它当做 当前持有锁的线程 可能是最好理解的
private transient volatile Node head;

// 阻塞的尾节点，每个新的节点进来，都插入到最后，也就形成了一个链表
private transient volatile Node tail;

// 这个是最重要的，代表当前锁的状态，0代表没有被占用，大于 0 代表有线程持有当前锁
// 这个值可以大于 1，是因为锁可以重入，每次重入都加上 1
private volatile int state;

// 代表当前持有独占锁的线程，举个最重要的使用例子，因为锁可以重入
// reentrantLock.lock()可以嵌套调用多次，所以每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer

大概了解了 aqs 底层的双向等待队列，
结构是这样的

每个 node 里面主要是的代码结构也比较简单

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
static final class Node {
    // 标识节点当前在共享模式下
    static final Node SHARED = new Node();
    // 标识节点当前在独占模式下
    static final Node EXCLUSIVE = null;

    // ======== 下面的几个int常量是给waitStatus用的 ===========
    /** waitStatus value to indicate thread has cancelled */
    // 代码此线程取消了争抢这个锁
    static final int CANCELLED =  1;
    /** waitStatus value to indicate successor's thread needs unparking */
    // 官方的描述是，其表示当前node的后继节点对应的线程需要被唤醒
    static final int SIGNAL    = -1;
    /** waitStatus value to indicate thread is waiting on condition */
    // 本文不分析condition，所以略过吧，下一篇文章会介绍这个
    static final int CONDITION = -2;
    /**
     * waitStatus value to indicate the next acquireShared should
     * unconditionally propagate
     */
    // 同样的不分析，略过吧
    static final int PROPAGATE = -3;
    // =====================================================


    // 取值为上面的1、-1、-2、-3，或者0(以后会讲到)
    // 这么理解，暂时只需要知道如果这个值 大于0 代表此线程取消了等待，
    //    ps: 半天抢不到锁，不抢了，ReentrantLock是可以指定timeouot的。。。
    volatile int waitStatus;
    // 前驱节点的引用
    volatile Node prev;
    // 后继节点的引用
    volatile Node next;
    // 这个就是线程本尊
    volatile Thread thread;

}

其实可以主要关注这个 waitStatus 因为这个是后面的节点给前面的节点设置的，等于-1 的时候代表后面有节点等待，需要去唤醒，
这里使用了一个变种的 CLH 队列实现，CLH 队列相关内容可以查看这篇 自旋锁、排队自旋锁、MCS锁、CLH锁

Posted on Edited on In Views: Disqus:

目前公司要对一些新的产品功能做灰度测试，因为在后端业务代码层面添加判断比较麻烦，所以想在nginx上做点手脚，就想到了openresty
前后也踩了不少坑，这边先写一点

首先是日志
error_log logs/error.log debug;
需要nginx开启日志的debug才能看到日志

使用 lua_code_cache off即可, 另外注意只有使用 content_by_lua_file 才会生效

1
2
3
4
5
6
7
8
http {
  lua_code_cache off;
}

location ~* /(\d+-.*)/api/orgunits/load_all(.*) {
   default_type 'application/json;charset=utf-8';
   content_by_lua_file /data/projects/xxx/current/lua/controller/load_data.lua;
}

使用lua给nginx请求response头添加内容可以用这个

1
ngx.header['response'] = 'header'

使用总结

后续：

  1. 一开始在本地环境的时候使用content_by_lua_file只关注了头，后来发到测试环境发现请求内容都没代理转发到后端服务上
    网上查了下发现content_by_lua_file是将请求的所有内容包括response都用这里面的lua脚本生成了，content这个词就表示是请求内容
    后来改成了access_by_lua_file就正常了，只是要去获取请求内容和修改响应头，并不是要完整的接管请求

  1. 后来又碰到了一个坑是nginx有个client_body_buffer_size的配置参数，nginx在32位和64位系统里有8K和16K两个默认值，当请求内容大于这两个值的时候，会把请求内容放到临时文件里，这个时候openresty里的ngx.req.get_post_args()就会报“failed to get post args: requesty body in temp file not supported”这个错误，将client_body_buffer_size这个参数配置调大一点就好了

  2. 还有就是lua的异常捕获，网上看一般是用pcall和xpcall来进行保护调用，因为问题主要出在cjson的decode，这里有两个解决方案，一个就是将cjson.decode使用pcall封装，

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    local decode = require("cjson").decode

    function json_decode( str )
        local ok, t = pcall(decode, str)
        if not ok then
          return nil
        end

        return t
    end

    这个是使用了pcall，称为保护调用，会在内部错误后返回两个参数，第一个是false，第二个是错误信息
    还有一种是使用cjson.safe包

    1
    2
    3
    4
    5
    6
    7
    local json = require("cjson.safe")
    local str = [[ {"key:"value"} ]]

    local t = json.decode(str)
    if t then
        ngx.say(" --> ", type(t))
    end

    cjson.safe包会在解析失败的时候返回nil

  3. 还有一个是redis链接时如果host使用的是域名的话会提示“failed to connect: no resolver defined to resolve “redis.xxxxxx.com””，这里需要使用nginx的resolver指令，
    resolver 8.8.8.8 valid=3600s;

  4. 还有一点补充下
    就是业务在使用redis的时候使用了db的特性，所以在lua访问redis的时候也需要执行db，这里lua的redis库也支持了这个特性，可以使用instance:select(config:get(‘db’))来切换db

  5. 性能优化tips
    建议是尽量少使用阶段钩子，例如content_by_lua_file，*_by_lua

  6. 发现一个不错的openresty站点
    地址

Posted on Edited on In Views: Disqus:

初识ambari

ambari是一个大数据平台的管理工具，包含了hadoop, yarn, hive, hbase, spark等大数据的基础架构和工具，简化了数据平台的搭建，之前只是在同事搭建好平台后的一些使用，这次有机会从头开始用ambari来搭建一个测试的数据平台，过程中也踩到不少坑，简单记录下。

简单过程

  • 第一个坑
    在刚开始是按照官网的指南，用maven构建，因为GFW的原因，导致反复失败等待，也就是这个guide，因为对maven不熟悉导致有些按图索骥，浪费了很多时间，之后才知道可以直接加repo用yum安装，然而用yum安装马上就出现了第二个坑。
  • 第二个坑
    因为在线的repo还是因为网络原因很慢很慢，用proxychains勉强把ambari-server本身安装好了，ambari.repo将这个放进/etc/yum.repos.d/路径下，然后yum update && yum install ambari-server安装即可，如果有条件就用proxychains走下代理。
  • 第三步
    安装好ambari-server后先执行ambari-server setup做一些初始化设置，其中包含了JDK路径的设置，数据库设置，设置好就OK了，然后执行ambari-server start启动服务，这里有个小插曲，因为ambari-server涉及到这么多服务，所以管理控制监控之类的模块是必不可少的，这部分可以在ambari-server的web ui界面安装，也可以命令行提前安装，这部分被称为HDF Management Pack，运行```ambari-server install-mpack \
  • -mpack=http://public-repo-1.hortonworks.com/HDF/centos7/2.x/updates/2.1.4.0/tars/hdf_ambari_mp/hdf-ambari-mpack-2.1.4.0-5.tar.gz \
  • -purge \
  • -verbose```
    安装，当然这个压缩包可以下载之后指到本地路径安装，然后就可以重启ambari-server

Posted on Edited on In Views: Disqus:

rabbitmq 介绍

接触了一下rabbitmq，原来在选型的时候是在rabbitmq跟kafka之间做选择，网上搜了一下之后发现kafka的优势在于吞吐量，而rabbitmq相对注重可靠性，因为应用在im上，需要保证消息不能丢失所以就暂时选定rabbitmq，
Message Queue的需求由来已久，80年代最早在金融交易中，高盛等公司采用Teknekron公司的产品，当时的Message queuing软件叫做：the information bus（TIB）。 TIB被电信和通讯公司采用，路透社收购了Teknekron公司。之后，IBM开发了MQSeries，微软开发了Microsoft Message Queue（MSMQ）。这些商业MQ供应商的问题是厂商锁定，价格高昂。2001年，Java Message queuing试图解决锁定和交互性的问题，但对应用来说反而更加麻烦了。
RabbitMQ采用Erlang语言开发。Erlang语言由Ericson设计，专门为开发concurrent和distribution系统的一种语言，在电信领域使用广泛。OTP（Open Telecom Platform）作为Erlang语言的一部分，包含了很多基于Erlang开发的中间件／库／工具，如mnesia／SASL，极大方便了Erlang应用的开发。OTP就类似于Python语言中众多的module，用户借助这些module可以很方便的开发应用。
于是2004年，摩根大通和iMatrix开始着手Advanced Message Queuing Protocol （AMQP）开放标准的开发。2006年，AMQP规范发布。2007年，Rabbit技术公司基于AMQP标准开发的RabbitMQ 1.0 发布。所有主要的编程语言均有与代理接口通讯的客户端库。

简单的使用经验

通俗的理解

这里介绍下其中的一些概念，connection表示和队列服务器的连接，一般情况下是tcp连接， channel表示通道，可以在一个连接上建立多个通道，这里主要是节省了tcp连接握手的成本，exchange可以理解成一个路由器，将消息推送给对应的队列queue，其实是像一个订阅的模式。

集群经验

rabbitmqctl stop这个是关闭rabbitmq，在搭建集群时候先关闭服务，然后使用rabbitmq-server -detached静默启动，这时候使用rabbitmqctl cluster_status查看集群状态，因为还没将节点加入集群，所以只能看到类似

1
2
3
4
Cluster status of node rabbit@rabbit1 ...
[{nodes,[{disc,[rabbit@rabbit1,rabbit@rabbit2,rabbit@rabbit3]}]},
 {running_nodes,[rabbit@rabbit2,rabbit@rabbit1]}]
...done.

然后就可以把当前节点加入集群，

1
2
3
4
5
6
7
rabbit2$ rabbitmqctl stop_app #这个stop_app与stop的区别是前者停的是rabbitmq应用，保留erlang节点，
                              #后者是停止了rabbitmq和erlang节点
Stopping node rabbit@rabbit2 ...done.
rabbit2$ rabbitmqctl join_cluster rabbit@rabbit1 #这里可以用--ram指定将当前节点作为内存节点加入集群
Clustering node rabbit@rabbit2 with [rabbit@rabbit1] ...done.
rabbit2$ rabbitmqctl start_app
Starting node rabbit@rabbit2 ...done.

其他可以参考官方文档

一些坑

消息丢失

这里碰到过一个坑，对于使用exchange来做消息路由的，会有一个情况，就是在routing_key没被订阅的时候，会将该条找不到路由对应的queue的消息丢掉What happens if we break our contract and send a message with one or four words, like "orange" or "quick.orange.male.rabbit"? Well, these messages won't match any bindings and will be lost.对应链接,而当使用空的exchange时，会保留消息，当出现消费者的时候就可以将收到之前生产者所推送的消息对应链接,这里就是用了空的exchange。

集群搭建

集群搭建的时候有个erlang vm生成的random cookie，这个是用来做集群之间认证的，相同的cookie才能连接，但是如果通过vim打开复制后在其他几点新建文件写入会多一个换行，导致集群建立是报错，所以这里最好使用scp等传输命令直接传输cookie文件，同时要注意下cookie的文件权限。
另外在集群搭建的时候如果更改过hostname，那么要把rabbitmq的数据库删除，否则启动后会马上挂掉

Posted on Edited on In Views: Disqus:

spark 的一些粗浅使用经验

工作中学习使用了一下Spark做数据分析,主要是用spark的python接口，首先是pyspark.SparkContext(appName=xxx)，这是初始化一个Spark应用实例或者说会话，不能重复，
返回的实例句柄就可以调用textFile(path)读取文本文件，这里的文本文件可以是HDFS上的文本文件，也可以普通文本文件，但是需要在Spark的所有集群上都存在，否则会
读取失败，parallelize则可以将python生成的集合数据读取后转换成rdd(A Resilient Distributed Dataset (RDD),一种spark下的基本抽象数据集),基于这个RDD就可以做
数据的流式计算，例如map reduce,在Spark中可以非常方便地实现

简单的mapreduce word count示例

1
2
3
4
textFile = sc.parallelize([(1,1), (2,1), (3,1), (4,1), (5,1),(1,1), (2,1), (3,1), (4,1), (5,1)])
data = textFile.reduceByKey(lambda x, y: x + y).collect()
for _ in data:
    print(_)

结果

1
2
3
4
5
(3, 2)
(1, 2)
(4, 2)
(2, 2)
(5, 2)