使用RethinkDB与SpringBoot 搭建一个简单的聊天

近项目有需要用到RethinkDB，故而在网上查询些资料，本文主要借鉴https://geowarin.github.io/spring-boot-and-rethinkdb.html 通过搭建一个聊天应用程序来学习RethinkDB.

RethinkDB 是什么？官网上的介绍是：RethinkDB is the open-source, scalable database that makes building realtime apps dramatically easier.

一、搭建rethinDB服务端

1、访问RethinkDB官网下载服务端：https://www.rethinkdb.com/docs/install/windows/ 我是在win7环境下搭建的，所以我下载的是windows版本的，如果各位是其他版本的，也可以在https://www.rethinkdb.com/docs/install/ 里面选择对应服务器的版本。


2、启动服务端：将下载好的rethinkdb-2.3.6.zip 解压后，双击rethinkdb.exe 及运行rethinkDB服务端了，默认相关数据是存储在当前目录的，文件名是rethinkdb_data。如果需要保存至指定目录是，需要用命令行执行如：C:\Users\Administrator\Desktop\rethinkdb-2.3.6>rethinkdb.exe -d f:\rethinkDB\data 前提是那个文件夹是存在的。 运行完毕以后，出现Server ready, 即代表服务启动完毕。打开浏览器：localhost:8080 看到如下dashboard 说明启动成功了

二、编写SpringBoot工程

1、增加maven依赖 主要包含springboot依赖包、org.webjars依赖包、以及RethinkDB的客户端依赖包

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>1.3.3.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
<version>1.3.3.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<version>1.3.3.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>1.3.3.RELEASE</version>
</dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>jquery</artifactId>
<version>3.0.0-alpha1</version>
</dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>sockjs-client</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>stomp-websocket</artifactId>
<version>2.3.3</version>
</dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>momentjs</artifactId>
<version>2.11.1</version>
</dependency>
<dependency>
<groupId>com.rethinkdb</groupId>
<artifactId>rethinkdb-driver</artifactId>
<version>2.3.3</version>
</dependency>
2、RethinkDB每个动作都是需要一个connection，所以先见一个工厂
public class RethinkDBConnectionFactory {
private String host;

private int port;

public RethinkDBConnectionFactory(String host,int port) {
this.host = host;
this.port = port;
}

public Connection createConnection() {
return RethinkDB.r.connection().hostname(host).port(port).connect();
}
}
配置类：

@Configuration
public class RethinkDBConfiguration {
// connect to docker
public static final String DBHOST = "127.0.0.1";
public static final int DBPORT = 28015;

@Bean
public RethinkDBConnectionFactory connectionFactory() {
return new RethinkDBConnectionFactory(DBHOST,DBPORT);
}

@Bean
DbInitializer dbInitializer() {
return new DbInitializer();
}
}

3、初始化DB，createDb 主要是rethinkDB的api，如果不熟悉可以，看看https://www.rethinkdb.com/docs/guide/java/ 里面的api入门使用
public class DbInitializer implements InitializingBean {
@Autowired
private RethinkDBConnectionFactory connectionFactory;

@Autowired
private ChatChangesListener chatChangesListener;

private static final RethinkDB r = RethinkDB.r;

public void afterPropertiesSet() throws Exception {
createDb();
chatChangesListener.pushChangesToWebSocket();
}

private void createDb() {
Connection connection = connectionFactory.createConnection();
List<String> dbList = r.dbList().run(connection);
if (!dbList.contains("chat")) {
r.dbCreate("chat").run(connection);
}
List<String> tables = r.db("chat").tableList().run(connection);
if (!tables.contains("messages")) {
r.db("chat").tableCreate("messages").run(connection);
r.db("chat").table("messages").indexCreate("time").run(connection);
}
}
}
4、聊天控制器

GET从DB中删除后20条消息

POST发出新消息

@RestController
@RequestMapping("/chat")
public class ChatController {

protected final Logger log = LoggerFactory.getLogger(ChatController.class);
private static final RethinkDB r = RethinkDB.r;

@Autowired
private RethinkDBConnectionFactory connectionFactory;

@RequestMapping(method = RequestMethod.POST)
public ChatMessage postMessage(@RequestBody ChatMessage chatMessage) {
chatMessage.setTime(OffsetDateTime.now());
HashMap run = r.db("chat").table("messages").insert(chatMessage)
.run(connectionFactory.createConnection());

log.info("Insert {}", run);
return chatMessage;
}

@RequestMapping(method = RequestMethod.GET)
public List<ChatMessage> getMessages() {

List<ChatMessage> messages = r.db("chat").table("messages")
.orderBy().optArg("index", r.desc("time"))
.limit(20)
.orderBy("time")
.run(connectionFactory.createConnection(), ChatMessage.class);

return messages;
}
}
ChatMessage 实体类
public class ChatMessage {
private String message;
private String from;
private OffsetDateTime time;

public ChatMessage() {
}

public ChatMessage(String message, String from, OffsetDateTime time) {
this.message = message;
this.from = from;
this.time = time;
}

public String getMessage() {
return message;
}

public void setMessage(String message) {
this.message = message;
}

public String getFrom() {
return from;
}

public void setFrom(String from) {
this.from = from;
}

public OffsetDateTime getTime() {
return time;
}

public void setTime(OffsetDateTime time) {
this.time = time;
}
}
监听数据库表的更新，会将更改的数据通过Websocket推送
@Service
public class ChatChangesListener {
protected final Logger log = LoggerFactory.getLogger(ChatChangesListener.class);

private static final RethinkDB r = RethinkDB.r;

@Autowired
private RethinkDBConnectionFactory connectionFactory;

@Autowired
private SimpMessagingTemplate webSocket;

@Async
public void pushChangesToWebSocket() {
Cursor<ChatMessage> cursor = r.db("chat").table("messages").changes()
.getField("new_val")
.run(connectionFactory.createConnection(), ChatMessage.class);

while (cursor.hasNext()) {
ChatMessage chatMessage = cursor.next();
log.info("New message: {}", chatMessage.getMessage());
webSocket.convertAndSend("/topic/messages", chatMessage);
}
}
}
配置Websocket
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

@Override
public void configureMessageBroker(MessageBrokerRegistry config) {
config.enableSimpleBroker("/topic");
}

public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/chatWS").withSockJS();
}
}

前端：index

<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Spring boot and Rethinkdb Chat</title>
<link rel="stylesheet" href="css/main.css">
</head>
<body>

<div id="container">
<div id="chat">
<ul id="messages">

</ul>
<form οnsubmit="sendMessage(); return false;">

<div class="chat-message clearfix">
<input name="message-to-send" autocomplete="off" id="messageInput" placeholder="Type your message" />

<button type="submit">Send</button>
</div>
</form>
</div>
</div>

<script src="webjars/jquery/3.0.0-alpha1/jquery.js"></script>
<script src="webjars/sockjs-client/1.0.0/sockjs.js"></script>
<script src="webjars/stomp-websocket/2.3.3/stomp.js"></script>
<script src="webjars/momentjs/2.11.1/moment.js"></script>
<script src="js/main.js"></script>
</body>
</html>
CSS样式main.css
*, *:before, *:after {
box-sizing: border-bo
}

ol, ul {
list-style: none;
}

body {
background: #C5DDEB;
font: 14px/20px "Lato", Arial, sans-serif;
padding: 40px 0;
color: white;
}

#container {
margin: 0 auto;
width: 750px;
background: #444753;
border-radius: 5px;
}

#chat {
width: 750px;
float: left;
background: #F2F5F8;
border-top-right-radius: 5px;
border-bottom-right-radius: 5px;
color: #434651;
}

#messages {
padding: 30px 30px 20px;
border-bottom: 2px solid white;
overflow-y: scroll;
height: 575px;
}

.message-data {
margin-bottom: 15px;
}

.message-data-time {
color: #92959E;
padding-left: 6px;
}

.message {
color: white;
padding: 10px;
font-size: 16px;
border-radius: 7px;
margin-bottom: 6px;
width: 90%;
position: relative;
}

.message:after {
bottom: ;
left: 7%;
content: " ";
height: 0;
width: 0;
position: absolute;
pointer-events: none;
border: 10px solid transparent;
border-bottom-color: #86BB71;
margin-left: -10px;
}

.my-message {
background: #86BB71;
}

.other-message {
background: #6C86D2;
}

.other-message:after {
border-bottom-color: #6C86D2;
left: 93%;
}

.chat-message {
padding: 30px;
}

#messageInput {
width: ;
border: none;
padding: 10px 20px;
font: 14px/22px "Lato", Arial, sans-serif;
margin-bottom: 10px;
border-radius: 5px;
resize: none;
}

button {
float: right;
color: #94C2ED;
font-size: 16px;
text-transform: uppercase;
border: none;
cursor: pointer;
font-weight: bold;
background: #F2F5F8;
}

button:hover {
color: #7fa9ce;
}

.align-left {
text-align: left;
}

.align-right {
text-align: right;
}

.float-right {
float: right;
}

.clearfix:after {
visibility: hidden;
display: block;
font-size: 0;
content: " ";
clear: both;
height: 0;
}
JS main.js
var userName = window.prompt("Enter your name", "some user");
//var userName = "lol";

function post(url, data) {
return $.ajax({
type: 'POST',
url: url,
headers: {
'Accept': 'application/json',
'Content-Type': 'application/json'
},
data: JSON.stringify(data)
})
}

function appendMessage(message) {
var time = message.time;
var fromNow = time.year +"-" + time.monthValue + "-" + time.dayOfMonth + " " + time.hour + ":" + time.minute + ":" +time.second;
var $message = $(`<li class="clearfix">
<div class="message-data ${message.from == userName ? 'align-left': 'align-right'}">
<span class="message-data-name">${message.from}</span>
<span class="message-data-time">${fromNow}</span>
</div>
<div class="message ${message.from == userName ? 'my-message': 'other-message float-right'}">
${message.message}
</div>
</li>`);
var $messages = $('#messages');
$messages.append($message);
$messages.scrollTop($messages.prop("scrollHeight"));
}

function getPreviousMessages() {
$.get('/chat').done(messages => messages.forEach(appendMessage));
}

function sendMessage() {
var $messageInput = $('#messageInput');
if($messageInput.val() ==null || $messageInput.val() ==""){
alert("不允许发空消息");
return;
}
var message = {message: $messageInput.val(), from: userName};
$messageInput.val('');
post('/chat', message);
}

function onNewMessage(result) {
var message = JSON.parse(result.body);
appendMessage(message);
}

function connectWebSocket() {
var socket = new SockJS('/chatWS');
stompClient = Stomp.over(socket);
//stompClient.debug = null;
stompClient.connect({}, (frame) => {
console.log('Connected: ' + frame);
stompClient.subscribe('/topic/messages', onNewMessage);
});
}

getPreviousMessages();
connectWebSocket();
resources的目录结构为：

三、启动客户端 必须使用jdk1.8

直接执行BootRethinkdbApplication的main方法，然后访问localhost:8093/index.html

具体的github的请见

https://github.com/StephenXiong/rethinkdb-boot