如果您了解用于通信的高流量应用程序、需要低延迟和良好容错能力的应用程序,您很可能已经遇到过 Elixir(作为一种编程语言)和 ScyllaDB(一种旨在低延迟的 NoSQL 数据库)的名称。两者的目标非常相似:处理通常需要更加关注稳定性的应用程序。
ScyllaDB 是全球公认的速度极快的数据库,它基于 Apache Cassandra,带来了多项低延迟改进。此外,ScyllaDB 是完全免费、开源的,并在 GNU AGPL 许可证下分发。
另一方面,Elixir 是一种编程语言,以很好地处理并发和容错等概念的适用性而闻名,这要归功于 Erlang 生态系统,在这种情况下,Elixir 使用名为 BEAM 的虚拟机,专为与大容量消息传递应用程序配合使用而设计。
本文的目的是介绍如何使用这两种技术创建您的第一个应用程序,通过这两种令人难以置信的技术进行开发,为您的未来打开大门和可能性。
事先我们需要安装 Elixir 和 ScyllaDB。我不会过多介绍安装的细节,因为这会使本文更简单。我们开始安装 Elixir。
一般来说,安装 Elixir 有两种主要方法:直接从包管理器安装或使用编程语言的版本管理器安装。就我而言,我将使用asdf
as 负责管理我的 Elixir 版本。如果您想使用包管理器安装 Elixir,请单击此处接收更多信息。
要安装,asdf
您可以单击此处进行检查。作为偏好,我总是选择使用“Bash & Git”或“Zsh & Git”进行安装。安装完成后,我们将准备安装以在我们的项目中接收最新版本的 Erlang 和 Elixir。安装 Erlang:
$ asdf plugin add erlang https://github.com/asdf-vm/asdf-erlang.git
$ asdf install erlang latest
$ asdf global erlang latest
您在安装时遇到任何问题吗?访问官方指南。
安装 Elixir:
$ asdf plugin-add elixir https://github.com/asdf-vm/asdf-elixir.git
$ asdf install elixir latest
$ asdf global elixir latest
您在安装时遇到任何问题吗?访问官方指南。
elixir -v
好吧,现在我们已经安装好了,我们可以通过在终端模拟器中输入来测试一切是否正常,我们将得到类似于以下内容的响应:
Erlang/OTP 25 [erts-13.2.2.2] [source] [64-bit]
Elixir 1.13.4 (compiled with Erlang/OTP 23)
这些是我安装的 Erlang 和 Elixir 版本。如果你想在你的机器上安装与我相同的版本,只需修改
asdf install
命令,替换latest
为版本号,然后将其设置asdf global elixir 1.13.4
为本例中的即可。
有多种使用 ScyllaDB 的方法,您可以将其与ScyllaDB Cloud一起使用,将 ScyllaDB安装在您的计算机上,或者按照我的使用方式:使用 Docker 容器。
如果您尚未安装 Docker,我建议您访问[安装]指南 ( https://docs.docker.com/engine/install/ )。如果你想使用ScyllaDB Cloud或安装在你的机器上的ScyllaDB,没有问题,只是在初始化和配置项目时小心放置节点链接,但是当我们到达这一部分时我会更好地解释它。
好吧,继续...要在 Docker 中使用 ScyllaDB 运行我们的容器,我们将使用以下命令:
$ docker run --name some-scylla -p 9042:9042 -d scylladb/scylla
该-p
选项表示我们希望将容器的端口 9042 与我们机器的端口 9042 绑定,从而允许我们的容器现在可以直接在我们的localhost:9042
.
为了测试连接,执行命令后,等待几秒钟,让容器中的所有内容正确启动,然后键入:
$ docker exec -it some-scylla cqlsh
因此您将看到类似以下内容的响应:
Connected to at 172.17.0.2:9042.
[cqlsh 5.0.1 | Cassandra 3.0.8 | CQL spec 3.3.1 | Native protocol v4]
Use HELP for help.
cqlsh>
这是我们的控制台,我们可以在其中执行命令来与 ScyllaDB 交互。默认情况下,使用的语言是CQL(Cassandra Query Language),与您可能已经接触过的标准数据库 SQL 非常相似。
好吧,让我们运行一个简单的命令来描述keyspaces
容器中的所有内容。keyspaces
可以用一个简单的类比来定义:当您使用像 MySQL 或 PostgreSQL 这样的关系数据库时,键空间基本上与数据库相同(定义有点超出这个范围,但我不会深入讨论)。
描述一下你的keyspaces
跑步:
cqlsh> DESCRIBE KEYSPACES;
您应该会看到类似以下内容的响应:
system_schema system_traces system_distributed
我们还没有创建任何东西keyspace
,对吧?好吧,让我们media_player
使用以下命令创建密钥空间,在本例中是用于 a 的:
cqlsh> CREATE KEYSPACE media_player
WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': '3'}
AND durable_writes = true;
让我们创建表:
cqlsh> CREATE TABLE media_player.songs (
id uuid,
titletext,
album text,
artist text,
created_at timestamp,
PRIMARY KEY (id, created_at)
);
要启动新项目,请运行命令:
$ mix new media_player
毕竟, Mix不仅仅是 Elixir 的依赖管理器,通过它我们可以运行和管理整个项目。默认情况下,Mix 已与 Elixir 一起安装。
将创建一个具有以下结构的项目:
.
├── README.md
├── lib
│ └── media_player.ex
├── mix.exs
好了,现在我们已经初始化了项目,我们可以开始玩了,所以打开您最喜欢的代码编辑器,然后开始吧。
此时,除了定义第一个设置之外,我们还将在 Elixir 中配置我们的项目,以安装和使用构建 CLI 所需的所有工具。
打开代码编辑器后,请注意有一个名为mix.exs
. 该文件负责定义有关我们项目的几个属性,包括在开发过程中将使用的依赖项。
稍微向下浏览页面,您将看到一个以defp deps do
... 开头的区域,正是在这部分中,我们将修改并插入以下依赖项:
# Run "mix help deps" to learn about dependencies.
defp deps
[
{:dotenv, "~> 3.0"},
{:decimal, "~> 1.0"},
{:xandra, "~> 0.14"},
{:elixir_uuid, "~> 1.2"}
]
end
好了,现在我们已经定义了依赖项,我们可以在终端模拟器中运行它:
$ mix deps.get
上面的命令将安装我们的
mix.exs
.
伟大的!现在我们可以在.env
.
好吧,现在让我们在项目的根目录创建一个名为的文件.env
(没错,与我们的 位于同一级别mix.exs
)。它将负责定义我们项目的第一个配置,包括将用于连接我们的集群的环境变量。
创建文件并在代码编辑器中打开它时,我们将定义:
SCYLLADB_USERNAME=
SCYLLADB_PASSWORD=
SCYLLADB_NODE=
SCYLLADB_KEYSPACE=
SCYLLADB_TABLE=
SCYLLADB_USERNAME
:配置为连接到 ScyllaDB 的用户名。SCYLLADB_PASSWORD
:为用户配置的密码。SCYLLADB_NODE
:连接到我们的节点的完整url,您可以只输入一个url(例如localhost:9042
),也可以定义生成的完整节点,并用逗号分隔(例如scylla-node1.com,scylla- node-2.com,scylla-node-3.com
)。SCYLLADB_KEYSPACE
:为我们的应用程序生成的密钥空间。SCYLLADB_TABLE
:将用于相应键空间的表。这样我们的 .env 应该看起来像:
SCYLLADB_USERNAME=scylla
SCYLLADB_PASSWORD=scylla
SCYLLADB_NODE=localhost:9042
SCYLLADB_KEYSPACE=media_player
SCYLLADB_TABLE=songs
完美的!现在我们的连接文件已经准备好了,我们可以开始构建项目了,好吗?
将连接模块分离在项目的单独区域中是优雅且有趣的,允许更实际的维护和有吸引力的组织,所以让我们创建两个带有路径的目录lib/media_player/config
。该目录将负责存储两个主要配置文件:用于连接集群的文件以及用于定义键空间和表的文件。
好吧,让我们在lib/media_player/config
目录中创建两个文件,分别称为connection.ex
和database.ex
。这样我们的目录结构现在将是:
.
├── README.md
├── lib
│ ├── media_player
│ │ └── config
│ │ ├── connection.ex
│ │ └── database.ex
│ └── media_player.ex
├── mix.exs
是的,我省略了该
deps
目录,因为它包含依赖项,也就是说,我们不会手动修改任何内容,不用担心,除了省略该目录之外,test
因为它将包含可以实现的测试,但是,不,我们暂时实施它。
好吧,现在我们可以从文件开始定义与集群的连接database.ex
。
defmodule MediaPlayer.Config.Database do
import Dotenv
load()
def start_link do
options = [
username: System.get_env("SCYLLADB_USERNAME"),
password: System.get_env("SCYLLADB_PASSWORD")
]
{:ok, cluster} =
Xandra.Cluster.start_link(
sync_connect: :infinity,
authentication: {Xandra.Authenticator.Password, options},
nodes:
System.get_env("SCYLLADB_NODE")
|> String.split(",")
)
cluster
end
end
在这个文件中我们:
Dotenv
库来管理文件中定义的变量.env
;load()
我们使用来自库的函数加载变量Dotenv
;start_link
,它将负责启动与我们的集群的连接链接;username
并password
从文件中接收这些值.env
;Xandra
我们使用来自 name 的函数初始化集群Xandra.Cluster.start_link
,负责启动与集群的连接链接sync_connect
value :infinity
!这意味着它将尝试以无限的预期响应时间建立连接(即,模块将等待必要的时间以完成所有节点的连接)。要了解更多信息,请点击此处;options
(之前定义的)作为参数;.env
文件加载并使用找到的逗号进行除法来定义节点,将它们分布在列表中(nodes
需要一个 url 列表来建立连接,这就是为什么需要分割来创建此列表)。如果您使用 ScyllaDB Cloud,这将使一切完美运行;完美的!我们的连接文件已准备就绪。现在让我们配置一个简单的区域,仅负责返回键空间和表,称为connection.ex
:
defmodule MediaPlayer.Config.Connection do
import Dotenv
load()
def keyspace() do
System.get_env("SCYLLADB_KEYSPACE")
end
def table() do
System.get_env("SCYLLADB_TABLE")
end
end
基本上,该模块的唯一功能是有两个函数来返回我们将使用的键空间和表,而无需始终使用该.env
库!
好吧,现在是另一个重要的点:由于我们的项目将有命令,因此,创建一个特定的模块来处理这些命令会很有趣,对吗?完美的!然而,在此之前,您认为创建一个模块来在数据库中执行查询怎么样,这样我们就可以集中执行查询的位置。
好吧,现在是时候创建一个lib/media_player
名为 的文件了actions.ex
。所以我们的目录结构将如下所示:
.
├── README.md
├── lib
│ ├── media_player
│ │ ├── actions.ex
│ │ └── config
│ │ ├── connection.ex
│ │ └── database.ex
│ └── media_player.ex
├── mix.exs
想知道!创建文件后,我们现在可以创建两个特定的函数,但为什么是两个呢?简单:该Xandra.Cluster.execute
函数有两种变体,第一个有两个参数(集群和要执行的查询),第二个有三个参数(集群、要执行的查询和参数,是一个列表,主要用于准备我们的查询)。
让我们开始吧,我们的模块应该如下所示:
defmodule MediaPlayer.Actions do
def cluster, do: MediaPlayer.Config.Database.start_link()
def run_query(query) do
case Xandra.Cluster.execute(cluster(), query) do
{:ok, result} ->
result
{:error, error} ->
IO.inspect(error)
end
end
def run_query(query, params) do
prepared = Xandra.Cluster.prepare!(cluster(), query)
case Xandra.Cluster.execute(cluster(), prepared, params) do
{:ok, result} ->
result
{:error, error} ->
IO.inspect(error)
end
end
end
在这个文件中我们:
cluster
函数,它除了初始化并将连接链接返回到集群之外什么也不做;run_query/1
一个只接受一个参数的函数(只是查询,毕竟集群已经是一个本地函数,我们知道我们总是会对其执行操作);Xandra.Cluster.execute
;:ok
,则表示一切顺利,那么我们返回完整的结果(查询的完整地图);:error
则说明执行查询时出现错误,那么我们检查具体的错误;
{:ok, result}
一个重要的细节是关于使用and进行的赋值{:error, error}
,因为在 Elixir 中一切都有一个返回,总是以一个原子开头的映射来验证给定的返回类型(我建议更多地了解原子),所以我们用映射绑定值!
run_query/2
带有两个参数的函数(只有查询和要执行的参数):Xandra.Cluster.execute
::ok
,则表示一切顺利,那么我们返回完整的结果(查询的完整地图);:error
则说明执行查询时出现错误,那么我们检查具体的错误;如果您想了解有关该Xandra.Cluster.execute
功能如何工作的更多信息,请单击此处。
您可能没有注意到的一个细节:两个函数具有相同的名称,但是它们的参数数量不同!这使得 Elixir 发挥了神奇的作用。函数是与它们期望接收的参数数量一起定义的,因此我将函数名称后跟斜杠“/”和参数数量。如果您想了解更多信息,请单击此处以更好地了解 Elixir 如何与模式匹配配合使用。
好吧,现在是期待已久的时刻:在我们的应用程序中添加负责执行命令的函数!为此,我们将创建一个lib/media_player
名为commands.ex
. 这样我们的目录结构将等于:
.
├── README.md
├── lib
│ ├── media_player
│ │ ├── actions.ex
│ │ ├── commands.ex
│ │ └── config
│ │ ├── connection.ex
│ │ └── database.ex
│ └── media_player.ex
├── mix.exs
让我们从创建整个模块的基础开始:
defmodule MediaPlayer.Commands do
alias MediaPlayer.Actions, as: Actions
alias MediaPlayer.Config.Connection, as: Connection
defp keyspace, do: Connection.keyspace()
defp table, do: Connection.table()
end
基本上上面我们定义了我们将有:
Actions
引用模块的别名MediaPlayer.Actions
;Connection
引用模块的别名MediaPlayer.Config.Connection
;keyspace
返回我们将使用的键空间值的私有函数;table
返回我们将使用的表值的私有函数;好吧,现在我们可以开始执行命令了,好吗?
好吧,这个命令将用于将歌曲添加到我们的数据库中。因此,我们将其分为两个主要函数,即add
和add_from
,它们接收四个要执行的参数。该add
函数只会收集要插入的数据:
defmodule MediaPlayer.Commands do
alias MediaPlayer.Actions, as: Actions
alias MediaPlayer.Config.Connection, as: Connection
defp keyspace, do: Connection.keyspace()
defp table, do: Connection.table()
def add_from(title, album, artist, created) do
query =
"INSERT INTO #{keyspace()}.#{table()} (id, title, album, artist, created_at) VALUES (?, ?, ?, ?, ?);"
{:ok, created, _} = DateTime.from_iso8601(created <> "T00:00:00Z")
Actions.run_query(query, [UUID.uuid4(), title, album, artist, created])
IO.puts("Song added!")
end
def add() do
title = IO.gets("Enter the title of the song: ") |> String.trim()
album = IO.gets("Enter the album of the song: ") |> String.trim()
artist = IO.gets("Enter the artist of the song: ") |> String.trim()
created =
IO.gets("Enter the date the song was created (YYYY-MM-DD): ")
|> String.trim()
add_from(title, album, artist, created)
end
end
嗯,这两个函数都非常具体,所以让我们分别举一个例子:
add
函数将:add_from
函数,将收集到的值作为参数传递;add_from/4
函数将:query
将要执行的;另一点:Elixir 中的“|>”管道的作用类似于“|” Unix shell 的管道,用于将函数的返回值作为下一个函数的第一个参数传递。在此处阅读有关管道的更多信息。
好了,现在我们有了负责添加已创建歌曲的功能!下次我们还去吗?
现在让我们创建一个函数,负责列出我们添加的所有歌曲。因此,我们将得到以下结果:
def list
query = "SELECT id, title, album, artist, created_at FROM #{keyspace()}.#{table()};"
Actions.run_query(query)
|> Enum.each(fn %{
"id" => id,
"title" => title,
"album" => album,
"artist" => artist,
"created_at" => created_at
} ->
IO.puts(
"ID: #{id} | Title: #{title} | Album: #{album} | Artist: #{artist} | Created At: #{created_at}"
)
end)
end
该list
函数不接收任何参数,毕竟它会打印添加到屏幕上的歌曲,如下所示:
id, title, album, artist, created_at
我们通过选择和 来定义查询,keyspace
并table
记住我们已经有两个返回这些值的函数!query
)并将其返回值传递给 an Enum.each
(类似于foreach
其他编程语言中的 a);一个重要的细节:在这种情况下,run_query
如果我们插入了多行,将返回多个值,对吧?好吧,run_query
返回一个完整的映射列表及其各自的值,因此我们用它Enum.each
来处理列表中的每个索引。
嗯,该delete
命令有点复杂,因为我们需要用户输入他想要删除的索引,并根据返回的索引列表对其进行验证!因此,实现的函数将导致如下结果:
def delete() do
query = "SELECT id, title, album, artist, created_at FROM #{keyspace()}.#{table()};"
songs =
Actions.run_query(query)
|> Enum.with_index(fn %{
"id" => id,
"title" => title,
"album" => album,
"artist" => artist,
"created_at" => created_at
},
index ->
IO.puts(
"Index: #{index + 1} | Title: #{title} | Album: #{album} | Artist: #{artist} | Created At: #{created_at}"
)
%{id: id, title: title, album: album, artist: artist, created_at: created_at}
end)
{input, _} = IO.gets("Enter the index of the song you want to delete: ") |> Integer.parse()
case Enum.at(songs, input - 1) do
%{} = song ->
query = "DELETE FROM #{keyspace()}.#{table()} WHERE id = ? AND created_at = ?;"
Actions.run_query(query, [song.id, song.created_at])
IO.puts("Song deleted!")
nil ->
IO.puts("Invalid index.")
end
end
好吧,基本上我们在这个函数中有:
query
;list
功能类似的Listing;Actions.run_query
并将其传递给Enum.with_index
索引;id
我们没有提供歌曲的歌曲名称,而是提供了一个手动索引(而不是从 0 开始,而是从 1 开始,因此index + 1
),用户将在其中键入(键入数字整数比 UUID 更实用,不是吗? );songs
;input
;songs
;Actions.run_query
并添加选项列表以作为参数绑定到查询;这样我们就完成了删除歌曲的功能了!
伟大的!必须操纵用户输入的命令已经完成!现在让我们创建一个额外的命令来负责对我们的数据库执行压力测试。我们将定义两个函数,即stress
初始化命令的函数,以及一个名称为 的私有函数,generate_stress_query
该函数接收名称为 的参数some_id
,并插入索引。
首先,我们必须在模块的开头添加以下部分:
defmodule MediaPlayer.Commands do
use Task
...
有了它Task
我们就可以进行异步调用,具有更好的实用性和性能。要了解更多信息,请点击此处。
因此,实现我们的功能:
defp generate_stress_query(some_id) do
current_date = Date.to_string(Date.utc_today())
"INSERT INTO #{keyspace()}.#{table()} (
id, title, album, artist, created_at
) VALUES (
#{UUID.uuid4()},
'Test Song #{some_id}',
'Test Artist #{some_id}',
'Test Album #{some_id}',
'#{current_date}'
);"
end
def stress
start = Time.utc_now()
cluster = MediaPlayer.Config.Database.start_link()
# Simple stress test
1..100_000
|> Task.async_stream(
fn id ->
IO.puts("[#{id}] Adding seed")
Xandra.Cluster.execute(cluster, generate_stress_query(id))
end,
max_concurrency: 500
)
|> Enum.to_list()
IO.puts("Time taken: #{Time.diff(Time.utc_now(), start, :second)} seconds")
end
基本上我们有:
stress
功能:Time.utc_now()
用;标记函数的开始时间id
函数Xandra.Cluster.execute
我们尝试通过调用函数来执行查询generate_stress_query
,该函数负责使用id
提供的内容生成完整的查询;max_concurrency
为500来限制异步调用的次数;generate_stress_query
功能:id
作为参数来生成查询;好了,这样我们的压力测试函数就准备好了!现在我们必须实现用户可以输入的命令输入!
MediaPlayer
让我们修改文件中的主模块media_player.ex
!好吧,首先让我们为命令定义一个别名:
defmodule MediaPlayer
alias MediaPlayer.Commands, as: Commands
end
别名将用于引用模块MediaPlayer.Commands
!
我们将实现的第一个函数是loop
,它将负责指挥命令、接收用户输入并保持在无限循环中始终等待输入,请参阅:
def loop do
IO.puts("-------------------------------------")
IO.puts("Type any command: ")
command = IO.gets("") |> String.trim()
case command do
"!add" ->
Commands.add()
loop()
"!list" ->
Commands.list()
loop()
"!delete" ->
Commands.delete()
loop()
"!stress" ->
Commands.stress()
loop()
"exit" ->
IO.puts("Bye bye!")
:OK
_ ->
IO.puts("Command not found!")
loop()
end
end
基本上我们等待用户输入命令并输入case
:
!add
调用该Commands.add()
函数;!list
调用该Commands.list()
函数;!delete
调用该Commands.delete()
函数;!stress
调用该Commands.stress()
函数;exit
打印一条再见消息并返回:ok
,定义该函数不再进入递归循环,结束我们的应用程序;_
打印命令未找到消息并进入递归循环,等待新输入;好了,现在我们的主要功能已经准备好了!但我们要如何执行呢?很简单,除了一个函数(将由初始函数调用的函数)之外,让我们创建一个start
负责启动应用程序的函数,请参阅:runstart
def start(_, _) do
run()
{:ok, self()}
end
def run do
IO.puts("-------------------------------------")
IO.puts("- ScyllaDB Cloud Elixir Media Player -")
IO.puts("- Leave a star on the repo -")
IO.puts("-------------------------------------")
IO.puts("Here some possibilities")
IO.puts(" !add - add new song")
IO.puts(" !list - list all songs")
IO.puts(" !delete - delete a specific song")
IO.puts(" !stress - stress testing with mocked data")
IO.puts("-------------------------------------")
loop()
end
基本上我们有:
start
功能:_
;run
函数;{:ok, self()}
定义该函数执行成功,关闭我们的应用程序;run
功能:loop
函数开始无限循环;我们完整的申请到此结束!现在我们要执行它吗?
好了,现在我们已经准备好了应用程序,让我们执行命令:
$ mix run
这样我们就可以开始与应用程序交互了!
如果您读到这里,非常感谢!本文的目的是演示如何将 ScyllaDB 与 Elixir 结合使用,并使用 Xandra 负责两者之间的连接进行简单演示。我强烈建议您关注ScyllaDB University的内容,以获得更多优质的教育内容,并查看有关使用 ScyllaDB 与DanielHe4rt和Cherry Ramatis一起使用的文章!两者都一直在贡献和发布精彩的内容。
我还建议您了解ScyllaDB Cloud 入门项目,该项目演示了 ScyllaDB 的一般用法,但是,更深入地研究如何使用 ScyllaDB 自己的平台来管理您的云集群,不是很酷吗?
在同一个存储库中,您也可以使用 Elixir 访问该项目,因此如果您想查看完整的源代码,请知道它位于上面提到的同一链接!
我希望您喜欢这些内容,并且希望它阐明了 ScyllaDB 与 Elixir 的用法。我也希望您有兴趣接收更多这样的内容或了解更多有关 Elixir 的信息。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。