事故

昨晚上准备把 sequences 服务扔到我的旧笔记本上然后写下单系统，然后发现一个很囧的事情……

消息序列化竟然失败了。

翻过去的代码，逐个排除问题，基本上结论是：因为过去我的测试写的太简单，所以没暴露出我写的基于 nippy 和 chesire 的序列器其实都是错的

因为我过去写的测试太简单，所以没正确的定位 kryo 序列化 Clojure 类型时遇到的问题

因为我过去写的测试太简单，没有正确的暴露出网络环境中不同操作系统的一些限制

因为我过去写的测试太简单，没有正确的暴露出 Akka remote 的一些配置约定

……

所以，测试还是很重很重要的呀`(~_~;)` 。

总结

本着能省事儿绝不折腾的精神，总结了几点体会：

网络绑定

惯例上我们会用 0.0.0.0 表示我们希望一个网络服务绑定环境内的所有地址，但是这个惯例在 Akka remote 里不成立 https://groups.google.com/forum/#!topic/akka-user/cRZmf8u\_vZY 。

Kryo 的对象管理

Kryo 的官方有给出一个建议，在项目开发的阶段打开 kryo-trace = true ，然后看哪些出现在 trace 日志中的类型没有被正确的分配id，而是给了一个较大的随机数，把这些类型都在 Init 代码中注册到 kryo 上，形如下例：

package liu.mars.market.serialization;

import clojure.lang.*;

import com.esotericsoftware.kryo.Kryo;

import com.fasterxml.jackson.databind.node.*;

import liu.mars.market.messages.CreateSequence;

import liu.mars.market.messages.DropSequence;

import liu.mars.market.messages.ListSequences;

import liu.mars.market.messages.NextValue;

import java.util.ArrayList;

import java.util.LinkedHashMap;

public class KryoInit {

public void customize(Kryo kryo){

kryo.register(CreateSequence.class);

kryo.register(DropSequence.class);

kryo.register(ListSequences.class);

kryo.register(NextValue.class);

kryo.register(ObjectNode.class);

kryo.register(ArrayNode.class);

kryo.register(LongNode.class);

kryo.register(TextNode.class);

kryo.register(ArrayList.class);

kryo.register(LinkedHashMap.class);

kryo.register(JsonNodeFactory.class);

kryo.register(PersistentArrayMap.class);

kryo.register(PersistentVector.class);

kryo.register(PersistentList.class);

kryo.register(Keyword.class);

kryo.register(Symbol.class);

kryo.register(LazySeq.class);

}

}

此时Sequences项目的server.conf如下：

akka {

extensions = ["com.romix.akka.serialization.kryo.KryoSerializationExtension$"]

actor {

provider = remote

serializers {

java = "akka.serialization.JavaSerializer"

kryo = "com.romix.akka.serialization.kryo.KryoSerializer"

}

serialization-bindings {

"liu.mars.market.messages.CreateSequence" = kryo

"liu.mars.market.messages.DropSequence" = kryo

"liu.mars.market.messages.ListSequences" = kryo

"liu.mars.market.messages.NextValue" = kryo

"com.fasterxml.jackson.databind.node.ObjectNode" = kryo

"com.fasterxml.jackson.databind.node.ArrayNode" = kryo

"clojure.lang.PersistentArrayMap" = kryo

"clojure.lang.PersistentList" = kryo

"clojure.lang.PersistentVector" = kryo

"clojure.lang.LazySeq" = kryo

"clojure.lang.Keyword" = kryo

"clojure.lang.Symbol" = kryo

"java.util.ArrayList" = kryo

"liu.mars.market.messages.Sequence" = kryo

}

kryo {

type = "graph"

idstrategy = "incremental"

buffer-size = 4096

max-buffer-size = -1

kryo-custom-serializer-init = "liu.mars.market.serialization.KryoInit"

kryo-trace = false

}

}

remote {

enabled-transports = ["akka.remote.netty.tcp"]

netty.tcp {

hostname = "192.168.50.22"

port = 25520

}

}

}

而 test.conf 如下：

akka {

extensions = ["com.romix.akka.serialization.kryo.KryoSerializationExtension$"]

actor {

provider = remote

serializers {

java = "akka.serialization.JavaSerializer"

kryo = "com.romix.akka.serialization.kryo.KryoSerializer"

nippy = "liu.mars.market.serialization.NippySerializer"

cheshire = "liu.mars.market.serialization.ChesireSerializer"

}

serialization-bindings {

"liu.mars.market.messages.CreateSequence" = kryo

"liu.mars.market.messages.DropSequence" = kryo

"liu.mars.market.messages.ListSequences" = kryo

"liu.mars.market.messages.NextValue" = kryo

"com.fasterxml.jackson.databind.node.ObjectNode" = kryo

"com.fasterxml.jackson.databind.node.ArrayNode" = kryo

"clojure.lang.PersistentArrayMap" = kryo

"clojure.lang.PersistentList" = kryo

"clojure.lang.PersistentVector" = kryo

"clojure.lang.LazySeq" = kryo

"clojure.lang.Keyword" = kryo

"clojure.lang.Symbol" = kryo

"java.util.ArrayList" = kryo

"liu.mars.market.messages.Sequence" = kryo

}

kryo {

type = "graph"

idstrategy = "incremental"

buffer-size = 4096

max-buffer-size = -1

kryo-custom-serializer-init = "liu.mars.market.serialization.KryoInit"

kryo-trace = true

}

}

remote {

enabled-transports = ["akka.remote.netty.tcp"]

// netty.tcp {

// hostname = "192.168.50.179"

port = 25521

// }

}

}

此时 test profile 的 config.edn 如下：

{:db-spec {:dbtype "postgresql"

:dbname "test"}

:remote {:sequences "akka.tcp://sequences@192.168.50.22:25520/user/sequences"}}

而 server profile 的 config.edn 如下：

{:db-spec {:dbtype "postgresql"

:dbname "sequences"

:user "market"

:password "market"}}

需要说明的是，这里的 ip 都是硬编码的我家里的局域网，读者们要运行示例的话需要自行编辑。数据库口令用了一个很弱的密码，这个在生产环境是要避免的。

另外 Ubuntu 桌面版默认不开放太大的端口号，而MacOS不开放太小的端口号，所以开发环境和测试环境的网络端口离的很远。

Clojure 对象的序列化

仔细看了一下 kryo 的报错，发现一个规律，它总是在反序列化Clojue的线性容器(vector/list/lazyseq 等等)的时候，提示无法反序列化 null。

这个消息么，让我想起一些很玄学的东西，比如 lisp 的 list 在语意上总有一个 nil 作为结尾……

但是现在走这条路就跑题太远了。

实际测试，Array 没有问题，ArrayList 没有问题，自定义的 POJO 也不会有问题，甚至 Jackson 的各种 Node 也都没有问题。

我在当前版本的 Sequences 服务中，就是把 LazySeq 用 into-array 变成数组，或者提取出其中的数据。总的来说避免Clojure List 类数据用 Kryo 来传递。

其实比较无脑的套路是用 Jackson 的 ObjectMapper 对象的 valueToTree 方法，经过实践是可以的。注意在开发过程中把前例没有加入的 Node 类型都注册到 kryo 就可以。

现在的 seq.clj 是这样的：

(nsliu.mars.market.seq

(:require [clojure.java.jdbc :as jdbc])

(:require [jaskell.sql :refer [selectfrom as]])

(:require [liu.mars.market.config :as config])

(:require [cheshire.core :refer [generate-string]]))

(defdb-spec (delay (config/db-spec)))

(defncreate-seq

[^String seq-name]

(->@db-spec

(jdbc/execute! (str"create sequence " seq-name))

first))

(defnlist-seq

[]

(let[sql (->(select["sequencename" :as :name " COALESCE(last_value, start_value)" as "last"] from :pg_sequences)

(.script))]

(->@db-spec

(jdbc/query [sql])

(into-array))))

(defnnext-val

[seq-name]

(->

(jdbc/query @db-spec ["select nextval(?)" seq-name])

first

(:nextval)))

(defndrop-seq

[seq-name]

(->@db-spec

(jdbc/execute! (str"drop sequence " seq-name))

first))

为了确保这次没有问题，我写了一个clojure的测试模块：

(nsliu.mars.market.remote-test

(:require [clojure.test :refer :all])

(:require [liu.mars.market.config :refer [conf]])

(:import (akka.testkit.javadsl TestKit)

(akka.actor ActorSystem)

(liu.mars.market.messages ListSequences CreateSequence NextValue DropSequence)

(java.util.function Supplier Function)))

(def^String url (->@conf

:remote

:sequences))

(testing "tests through akka tcp provider"

(let[system (ActorSystem/create "test")

test-kit (TestKit. system)

self (.getRef test-kit)

await#(.awaitCond test-kit (reify Supplier (get[this] (.msgAvailable test-kit))))

remote (.actorSelection system url)

seq-name "test"]

(testing "tests about a sequence lifetime"

(.tell remote (ListSequences.) self)

(await)

(.expectMsgPF test-kit "message should't include test"

(reify Function

(apply[this msg]

(let[data (vec msg)]

(is (not-any?#(=(:name %) seq-name) data))))))

(.tell remote

(doto(CreateSequence.)

(.setName seq-name))

self)

(await)

(.expectMsgClass test-kit Integer)

(.tell remote (ListSequences.) self)

(await)

(.expectMsgPF test-kit "message should include test"

(reify Function

(apply[this msg]

(let[data (vec msg)]

(is (some#(=(:name %) seq-name) data))))))

(dotimes[n 100]

(.tell remote (doto(NextValue.) (.setName "test")) self)

(await)

(.expectMsgPF test-kit "should get next value more than times"

(reify Function

(apply[this msg]

(is (=(incn) msg))))))

(.tell remote (doto(DropSequence.) (.setName seq-name)) self)

(await)

(.expectMsgClass test-kit Integer)

(.tell remote (ListSequences.) self)

(await)

(.expectMsgPF test-kit "message should't include test"

(reify Function

(apply[this msg]

(let[data (vec msg)]

(is (not-any?#(=(:name %) seq-name) data))))))

(TestKit/shutdownActorSystem system))))

代码风格不算太干净，简单的实验了一遍构造一个 sequence 然后调用最后删除的过程。