假设有一个热门的国际购物平台,它需要处理高并发的购物订单。因为它是为世界各地的用户设计,它应该能够支持不同的货币结算。当用户购买商品时,系统会根据当前汇率将原价格兑换成目标货币的价格。
基于4台云服务器,使用Zookeeper, Kafka, Spark服务框架和MySQL,设计并实现一个分布式交易结算系统,功能包括接收和处理贸易订单、记录所有交易结果和总交易金额、定时更新汇率,在实现基本功能的基础上尽可能地优化throughput和latency、支持高并发。
- centos
- 8GB DRAM * 4
- 4-core CPU * 4
- 各服务器的职能
虚拟机 | zookeeper | kafka | spark |
---|---|---|---|
dist-1 | ✔ | ✔ | master |
dist-2 | ✔ | ✔ | slave |
dist-3 | ✔ | ✔ | slave |
dist-4 | slave |
首先需要下载Zookeeper, Kafka, Spark等各种包,因此需要先安装wget指令
yum -y install wget
配置四台虚拟机之间的域名映射
# /etc/hosts
10.0.0.43 dist-1
10.0.0.18 dist-2
10.0.0.86 dist-3
10.0.0.47 dist-4
首先下载合适版本的包,这里选择了zookeeper-3.4.14.tar.gz
wget https://mirrors.tuna.tsinghua.edu.cnlog4j/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
解压至指定文件夹
tar zxvf zookeeper-3.4.14.tar.gz -C ~/soft
创建软链接方便使用
cd ~/soft
ln -s zookeeper-3.4.14 zk
随后修改默认的配置文件
cd conf
cp zoo_sample.cfg zoo.cfg
zoo_sample.cfg为里面自带的样例配置文件,这里直接采用它,需要修改一下:
# ~/soft/zk/conf/zoo.cfg
dataDir=/home/centos/zookeeper/data
server.1=dist-1:2888:3888
server.2=dist-2:2888:3888
server.3=dist-3:2888:3888
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.1/kafka_2.11-2.2.1.tgz
tar zxvf kafka_2.11-2.2.1.tgz -C ~/soft
cd ~/soft
ln -s kafka_2.11-2.2.1.tgz kafka
修改配置文件
# ~/soft/kafka/config/server.properties
broker.id=0 # different in each node
zookeeper.connect=dist-1:2181,dist-2:2181,dist-3:2181 #zookeeper config
spark与hadoop的关系
- spark使用hdfs作为分布式的文件系统,而在local或者standalone模式下不需要hdfs,因此不需要先安装hadoop
wget http://mirrors.tuna.tsinghua.edu.cn/apache/spark/spark-2.4.3/spark-2.4.3-bin-hadoop2.7.tgz
tar zxvf spark-2.4.3-bin-hadoop2.7.tgz -C ~/soft
cd ~/soft
ln -s spark-2.4.3-bin-hadoop2.7.tgz spark
修改配置文件
# ~/soft/spark/conf/slaves
dist-1
dist-2
dist-3
在三台机器上都配置slaves文件
选择使用dist-1作为数据库服务器,本地配置mysql。
# 下载mysql的repo源
> wget http://repo.mysql.com/mysql-community-release-el7-5.noarch.rpm
# 安装mysql-community-release-el7-5.noarch.rpm包
> sudo rpm -ivh mysql-community-release-el7-5.noarch.rpm
# 安装mysql,并按照命令行提示设置密码
> sudo yum install mysql-server -y
# 登录mysql
> mysql -u root -p
# 创建数据库lab5和表格commodity, result
create database lab5;
create table commodity(
id INT NOT NULL PRIMARY KEY,
name VARCHAR(11) NULL,
price DOUBLE NULL,
currency VARCHAR(5) NULL,
inventory INT NULL) ENGINE = InnoDB;
create table result(
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NULL,
initiator VARCHAR(5) NULL,
success VARCHAR(5) NULL,
paid DOUBLE NULL) ENGINE = InnoDB;
# your/path/to/zookeeper/bin
./zkServer.sh start
# your/path/to/kafka/bin
./kafka-server-start.sh ../config/server.properties
spark启动前需要在master节点上配置slaves文件
# your/path/to/spark/conf/slaves
[worker1-hostname]
[worker2-hostname]
[worker3-hostname]
...
启动服务
# your/path/to/spark/sbin
./start-all.sh
这个脚本会自动启动各个slave上的worker,不然需要在各个slaves上各自启动worker
提交Spark应用
spark-submit --master spark://[master-node]:7077 [yourapp].jar
kafka提供了许多简易的API可以直接调用,使用kafka.producer API实例化生产者。
Web/WebApp.java: 向kafka集群发送订单数据的producer。
override NanoHTTPD.Response serve函数,使web server有如下功能:
功能 | web url |
---|---|
清空数据库 | /clear/database |
发送订单请求 | /post/order |
向数据库中插入新商品 | /insert/commodity |
获取结果 | /get/result |
获取事务总量 | /get/totalTxAmount |
-
使用
java.util.Properties
配置并初始化kafka producer实例。- 加入集群节点到broker-list
- 发送的message类型为String,所以设置序列化参数为StringEncoder
-
调用
producer.send()
接口,将缓冲池中的消息异步地发送到broker的指定topic中。异步发送是指,
send()
将消息存储到底层待发送的I/O buffer后,将立即返回,从而并行无阻塞地发送更多消息,因此kafka能批量处理消息(batch)以提高效率。
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public WebApp(int port) throws IOException{
super(port);
Properties properties = new Properties(); //--2
properties.put("metadata.broker.list","dist-1:9092,dist-2:9092,dist-3:9092");
properties.put("serializer.class","kafka.serializer.StringEncoder");
properties.put("request.require.acks","1");
ProducerConfig config=new ProducerConfig(properties);
producer=new Producer<>(config);
}
static public void main(String[] args){
try {
new WebApp(30441).start(NanoHTTPD.SOCKET_READ_TIMEOUT, false);
} catch (IOException e) {
e.printStackTrace();
}
}
该系统通过Spark Streaming读取Kafka Producer的消息,转换成可操作的、按照时间段分割的数据流,用批处理的方式处理订单。
App.java:
-
通过 JavaStreamingContextFactory构建Streaming context对象,指明应用名称"Order Processing"、时间窗口大小(即批处理时间间隔)为3s 。
SparkConf conf = new SparkConf().setAppName("Order Processing"); JavaSparkContext sc = new JavaSparkContext(conf); sc.setLogLevel("WARN"); JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(2));
-
创建inputDstream,定义数据源。本项目利用KafkaStream的API,创建Kafka topic后,直接读取kafka。
Map<String, Integer> topicMap = new HashMap<String, Integer>(); topicMap.put("kafka_spark", 1); JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createDirectStream(jssc, "dist-1:2181,dist-2:2181,dist-3:2181", "spark_receiver", topicMap);
-
对messages进行map操作按时间切分、转换成DStream,再进行map操作传入订单处理模块,进行处理返回结果的DStream。
JavaDStream<String> results = lines.map(OrderProcessor::process);
-
DStream: 是Spark Streaming中的一个基本抽象,代表数据流,隐藏了实现细节。DStream可以从kafka等输入源获得,也可以转换得到。在 DStream 内部维护了一组离散的以时间轴为键的 RDD 序列,每个RDD 包含了指定时间段内的数据流,我们对于 DStream 的各种操作最终都会映射到内部的 RDD 上,最终提交给Spark处理。
-
配置后调用start()正式启动Spark Streaming。
jssc.start(); jssc.awaitTermination();
init(): 初始化Zookeeper服务器,若服务器上不存在 /lock
节点则创建,该节点是一个持久节点(PERSISTENT)
Stat stat = zookeeper.exists(parentPath, null);
if (stat == null) {
zookeeper.create(parentPath, "for lock".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
lock():
- 在父节点
/lock
下创建一个临时顺序子节点(EPHEMERAL_SEQUENTIAL),该节点会在客户端断开连接时删除,并且服务器会给该节点加上一个全局唯一的顺序后缀。指定子节点的前缀,最终创建的节点路径为/lock/lock-0000000001
.
String lockPath = zookeeper.create(lockPrefix, "lock node".getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
- 获取
/lock
节点的所有子节点,按后缀排序后判断第一步创建的节点是不是所有子节点中顺序最小的节点,如果是,则lock()函数返回,调用者获得锁
List<String> children = zookeeper.getChildren("/lock", false);
Collections.sort(children, new Comparator<String>() {
public int compare(String left, String right) {
String leftId = left.substring(lockPrefix.length());
String rightId = right.substring(lockPrefix.length());
return leftId.compareTo(rightId);
}
});
if (nodePath.equals(children.get(0))) {
System.out.println(nodePath + " acquire the lock");
}
- 若当前节点不是最小的节点,则创建
LockWatcher
来监听前一个节点的删除事件,接着LockWatcher
会调用latch.countDown()
使得正在等待latch
的lock()
函数能够继续执行下去获得锁,由于分布式并发场景下有可能在一个结点await前先前结点先被delete,造成死锁,因此增加超时重查机制
CountDownLatch latch = new CountDownLatch(1);
LockWatcher lockWatcher = new LockWatcher(latch);
Stat stat = zookeeper.exists(prePath, lockWatcher);
if (stat != null) {
while(!localLatch.await(200,TimeUnit.MILLISECONDS)){
stat = zookeeper.exists(prePath, false);
if(stat == null) break;
}
}
unlock(): 删除自己取得的lockPath
锁节点
结合3.2与3.3.1的部分,Spark Streaming从Kafka producer接收到order processing任务后,从master分发给合适的slave,slave中的worker获取zookeeper的锁之后可以访问MySQL数据库,关系见如下示意图:
main(): 定义4个并行的线程,分别对应4种货币,每分钟修改1次货币汇率。
static public void main(String[] args) {
String[] currencies = {"RMB","USD","JPY","EUR"};
Double[] initValues = {2.0, 12.0, 0.15, 9.0};
CurrentChange[] threads = new CurrentChange[currencies.length];
for (int i = 0; i < threads.length; i++) {
threads[i] = new CurrentChange(currencies[i],initValues[i]);
}
for (CurrentChange thread : threads) {
thread.start();
}
}
CurrentChange类实现 :继承Java.Thread类,@Override重写Thread.run()
方法,使调用代码更简洁。
public class CurrentChange extends Thread {
@Override
public void run(){
//implementation of changing exchange rate every 60s
}
}
- MySQL位于dist-1上,集群通过hibernate配置连接3306端口的数据库。
- Result的id设置为AUTO_INCREMENT自增。
- 此项目中使用zookeeper实现分布式锁作为应用层的锁,因此在数据库层面不需要增加隔离。
order json: input-0.json, input-1.json, input-2.json,约2w条数据。
LockTest.java: 用于测试zookeeper锁实现的正确性、可扩展性。
分析latency与throughput:
-
Latency: 由于kafka的高性能,latency主要来自于Spark Streaming自身的流处理中,由于Spark Streaming采用batch的方式,并不是来一条处理一条的真实时处理(如Storm),因此latency主要取决于process time以及batch interval,因此在这将latency视为单个Record处理的时间的平均值,可以通过Spark UI查看得到。 $$ latency ≈ 0.5 * Batch\ Interval + processtime $$
-
Throughput: 由于应用process time相对较长,因此单个Spark Receiver足以满足任务的吞吐量需求,因此主要瓶颈仍然在于process time。
-
总结:优化重点在于减少process time,所以优化需要进一步分析任务处理时间。
按照不同的事务级别、锁实现、forceSync设置,采用控制变量法进行throughput的测试,处理2w条order:
MySQL事务级别 | zookeeper lock实现 | forceSync | Throughput |
---|---|---|---|
Read Uncommitted | 无 | 250.0 | |
Read Repeatable | 无 | 243.9 | |
Read Uncommitted | 无 | ✔ | 125 |
Read Repeatable | 无 | ✔ | 108.7 |
Read Uncommitted | commodity lock | ✔ | 11.1 |
Read Uncommitted | commodity lock | 139.1 | |
Read Repeatable | commodity lock | 137 | |
Read Repeatable | 全局single lock | 61 | |
Read Uncommitted | 全局single lock | 62.1 | |
Read Uncommitted | 全局single lock | ✔ | 19 |
ps: 由于latency不好测量且取决于Batch Interval,对Spark应用来说并不追求精确到秒级别的latency,因此这里选择Throughput作为测试指标。
ps2: 由于在开启forceSync模式下overhead主要来自于zookeeper的操作开销,数据库事务隔离级别影响可以忽略,因此仅测试了Read Repeatable。
由于对整个表加了锁,相当于每个订单都串行执行,因此虽然通过Spark启动多个Executor并行处理多个Task,却没有利用到并行,由于应用只会涉及到更新自己订单的商品,因此只要对每个商品单独加锁,在没有冲突的情况下就可以并行处理。
在前一个版本的基础上实现了对商品单独加锁
可以观察到Throughput反而降低,经过检查发现虽然加入商品锁之后各个任务之间可以并行,但由于任务处理时间大部分都是zookeeper的写入操作(包括create以及delete),因此商品锁相较于单个全局锁多了几倍的overhead,throughput反而下降。
以下是运行时间分析图,Throughput单位order/second,时间单位ms:
可以看到在Single Lock模式下加锁解锁时间相对较短(加锁时间长是由于将线程等待时间也算进去了),平均操作只占20-30ms,这点也可以从更新Tx的时间上得到验证,加锁解锁与更新Tx都相当于一个zookeeper写入操作,而在两种模式下更新Tx的时间都相当稳定为20-30ms。
在前一个版本的基础上配置zoo.cfg
# your/path/tp/zookeeper/conf/zoo.cfg
forceSync=no
这个参数的作用是强制将zookeeper的写入操作持久化并且保持顺序,因此如果有多个写入操作并发,后面的写入操作需要等待前面的写入forceSync完之后才能forceSync并且返回,造成latency增加。
可以观察到Throughput大幅度增加,tradeoff为zookeeper的可靠性降低,如果一个节点崩溃重启可能无法从本地数据recover,但我们采用三台作为一个集群,如果只有一台崩溃可以从其他两台上恢复数据,因此可靠性仍然可以保证。
将Mysql的事务隔离级别降到最低,因为我们已经在应用层实现了锁,因此不需要数据库层面为我们提供隔离。
Throughput有一定提升但有限,因为数据库操作本身占用时间便不大,并且在我们的应用中将每一次读或写操作单独作为一个事务,因此事务之间的冲突本身就不多。
Q1: kafka-console-consumer.sh --zookeeper xxx 报错
A: 因为版本更新该参数改为--bootstrap-server,需要broker server而不是zookeeper server
Q2: zkServer.sh start后status显示not running
A: 可查看zookeeper.out文件
org.apache.zookeeper.server.quorum.QuorumPeerConfig$ConfigException: Error processing /home/centos/soft/zk/bin/../conf/zoo.cfg
at org.apache.zookeeper.server.quorum.QuorumPeerConfig.parse(QuorumPeerConfig.java:156)
at org.apache.zookeeper.server.quorum.QuorumPeerMain.initializeAndRun(QuorumPeerMain.java:104)
at org.apache.zookeeper.server.quorum.QuorumPeerMain.main(QuorumPeerMain.java:81)
Caused by: java.lang.IllegalArgumentException: /home/centos/zookeeper/data/myid file is missing
at org.apache.zookeeper.server.quorum.QuorumPeerConfig.parseProperties(QuorumPeerConfig.java:408)
at org.apache.zookeeper.server.quorum.QuorumPeerConfig.parse(QuorumPeerConfig.java:152)
... 2 more
- 由于dataDir下的myid文件未创建
- 若日志显示正常status却未显示,可能由于集群模式还未完成选举,等所有机器都启动后再查看
Q3: Field "id" doesn't have a default value
A: 由于使用hibernete将Result表的id列设置为@GeneratedValue(strategy = GenerationType.IDENTITY)
因此自增属性交由Mysql管理,而生产环境下的Mysql未配置id为AUTO INCREMENT,因此报错,通过alter table Result modify id int AUTO INCREMENT;
修改完毕,需要保证连接数据库的进程关闭,否则会卡死
Q4: 发现spark应用消费速度过慢,只有个位数throughput
A: 首先排查kafka本身吞吐量,通过kafka-producer-perf-test.sh测试发现kafka的吞吐量正常,其次怀疑任务本身过于耗时,通过不执行任务直接将input的数据print出来,问题没有得到解决,发现由于Dstream的print方法在数量大于10个时后续以省略号表示
Q5: Web请求处理速度过慢,throughput仅有十几
A: 由于之前在本机上发送订单请求,怀疑由于Web Receiver瓶颈,因此将sender的python脚本进行打包,在集群上进行send,打算采用多个Receiver方式,然后发现服务器上send速度很快,因此问题为开发机至服务器间的网络
Q6: Zookeeper产生死锁
A:由于共享static变量, 多个worker/多线程拿锁产生问题,没有有效放锁。修改lockService类的实现,删去lockPath的static变量,并且每次zookeeper删除节点时都删除最小节点
Q7: 在master节点上使用start-all.sh但Worker没有启动
A: 由于master默认采用主机名作为连接地址,而openstack主机名采用[实例名].novalocal的格式,因此Worker启动无法找到master而启动失败
学号 | 姓名 | 分工 |
---|---|---|
516030910328 | 蔡忠玮 | 环境搭建、Spark应用、testfile |
516030910219 | 徐家辉 | zookeeper分布式锁、订单处理应用 |
516030910422 | 赵樱 | Web应用 |
516030910367 | 应邦豪 | 更改汇率应用 |
项目Github:https://github.com/sansazhao/Distributed-System-group7
项目结构及说明:
├─Core
│ CommodityService.java
│ ResultService.java
│ CurrentService.java zookeeper管理汇率表
│ LockService.java zookeeper分布式锁实现
│ Processor.java 订单处理
│ HibernateUtil.java 数据库连接配置
├─Current
│ CurrentApp.java
│ CurrentChange.java 继承thread,修改汇率的实现
├─Entity
│ Commodity.java
│ Result.java
├─Spark
│ SparkApp.java
│ OrderProcessor.java 测试使用
│ OrderProducer.java 测试使用
└─Web
WebApp.java kafka producer(发送订单、添加商品)