Elixir/Phoenix 2025 - 高並行リアルタイムWeb開発

Elixir/Phoenixとは

Elixirは、Erlang VM (BEAM) 上で動作する関数型プログラミング言語です。Phoenixは、Elixirで書かれた高性能Webフレームワークで、リアルタイムWebアプリケーションの構築に特化しています。2025年現在、Discord、Pinterest、PepsiCoなど多くの大規模サービスで採用され、その信頼性と拡張性が実証されています。

注目ポイント: Discordは1つのサーバーで500万以上の同時接続を処理しており、Elixir/Phoenixの圧倒的なスケーラビリティを証明しています。

flowchart TB
 subgraph BEAM["BEAM VM (Erlang Virtual Machine)"]
 subgraph Scheduler1["Scheduler 1"]
 P1["Process"]
 P2["Process"]
 end
 subgraph Scheduler2["Scheduler 2"]
 P3["Process"]
 P4["Process"]
 end
 subgraph SchedulerN["Scheduler N"]
 P5["Process"]
 P6["Process"]
 end
 end

 subgraph OTP["OTP (Open Telecom Platform)"]
 GenServer["GenServer"]
 Supervisor["Supervisor"]
 Application["Application"]
 end

 subgraph Phoenix["Phoenix Framework"]
 LiveView["LiveView"]
 Channels["Channels"]
 Ecto["Ecto"]
 PubSub["PubSub"]
 end

 BEAM --> OTP
 OTP --> Phoenix

BEAM VMの特徴

軽量プロセスと並行性

BEAM VMの最大の特徴は、軽量プロセスによる並行処理モデルです。OSスレッドとは異なり、各プロセスはわずか数KBのメモリしか消費せず、数百万のプロセスを同時に実行できます。

# 100万のプロセスを起動する例
defmodule MassiveProcesses do
 def start_processes(count) do
 1..count
 |> Enum.map(fn i ->
 spawn(fn ->
 # 各プロセスは独立して動作
 receive do
 {:message, msg} -> IO.puts("Process #{i}: #{msg}")
 end
 end)
 end)
 end

 def send_to_all(pids, message) do
 Enum.each(pids, fn pid ->
 send(pid, {:message, message})
 end)
 end
end

# 使用例
pids = MassiveProcesses.start_processes(1_000_000)
MassiveProcesses.send_to_all(pids, "Hello!")

プロセス間通信

Elixirのプロセスは共有メモリを持たず、メッセージパッシングで通信します。これにより、デッドロックやレースコンディションのリスクが大幅に軽減されます。

defmodule Counter do
 use GenServer

 # クライアントAPI
 def start_link(initial_value \\ 0) do
 GenServer.start_link(__MODULE__, initial_value, name: __MODULE__)
 end

 def increment do
 GenServer.call(__MODULE__, :increment)
 end

 def get_value do
 GenServer.call(__MODULE__, :get_value)
 end

 # サーバーコールバック
 @impl true
 def init(initial_value) do
 {:ok, initial_value}
 end

 @impl true
 def handle_call(:increment, _from, state) do
 {:reply, state + 1, state + 1}
 end

 @impl true
 def handle_call(:get_value, _from, state) do
 {:reply, state, state}
 end
end

OTPによる耐障害性設計

Supervisorツリー

OTP (Open Telecom Platform) は、Erlang/Elixirの強力なライブラリ群です。特にSupervisorパターンは「Let it crash」の思想に基づき、障害から自動的に回復するシステムを構築できます。

defmodule MyApp.Application do
 use Application

 @impl true
 def start(_type, _args) do
 children = [
 # データベース接続プール
 {MyApp.Repo, []},
 # PubSubシステム
 {Phoenix.PubSub, name: MyApp.PubSub},
 # Webサーバー
 MyAppWeb.Endpoint,
 # カスタムワーカー（最大3回再起動）
 %{
 id: MyApp.CriticalWorker,
 start: {MyApp.CriticalWorker, :start_link, []},
 restart: :permanent,
 max_restarts: 3,
 max_seconds: 5
 }
 ]

 opts = [strategy: :one_for_one, name: MyApp.Supervisor]
 Supervisor.start_link(children, opts)
 end
end

flowchart TB
 subgraph SupervisorTree["Supervisor Tree"]
 AppSup["Application Supervisor<br/>(one_for_one)"]

 subgraph Workers["Workers"]
 Repo["Repo<br/>(DB Pool)"]
 PubSub["PubSub"]
 Endpoint["Endpoint"]
 Worker["CriticalWorker"]
 end

 AppSup --> Repo
 AppSup --> PubSub
 AppSup --> Endpoint
 AppSup --> Worker
 end

 Crash["Worker Crash"] --> Worker
 Worker -.-> |"Auto Restart"| Worker

再起動戦略

Phoenix LiveView

リアルタイムUIの革新

Phoenix LiveViewは、サーバーサイドレンダリングとWebSocketを組み合わせ、JavaScriptをほとんど書かずにリアルタイムUIを構築できる革新的な機能です。

defmodule MyAppWeb.CounterLive do
 use MyAppWeb, :live_view

 @impl true
 def mount(_params, _session, socket) do
 if connected?(socket) do
 # クライアント接続後にリアルタイム更新を開始
 :timer.send_interval(1000, self(), :tick)
 end

 {:ok, assign(socket, count: 0, last_updated: DateTime.utc_now())}
 end

 @impl true
 def handle_event("increment", _params, socket) do
 {:noreply, update(socket, :count, &(&1 + 1))}
 end

 @impl true
 def handle_event("decrement", _params, socket) do
 {:noreply, update(socket, :count, &(&1 - 1))}
 end

 @impl true
 def handle_info(:tick, socket) do
 {:noreply, assign(socket, last_updated: DateTime.utc_now())}
 end

 @impl true
 def render(assigns) do
 ~H"""
 <div class="counter-container">
 <h1>リアルタイムカウンター</h1>

 <div class="counter-value">
 <%= @count %>
 </div>

 <div class="counter-controls">
 <button phx-click="decrement" class="btn btn-danger">
 -
 </button>
 <button phx-click="increment" class="btn btn-primary">
 +
 </button>
 </div>

 <p class="last-updated">
 最終更新: <%= Calendar.strftime(@last_updated, "%H:%M:%S") %>
 </p>
 </div>
 """
 end
end

LiveViewのリアルタイムチャット実装

より実践的な例として、リアルタイムチャットアプリケーションを実装してみましょう。

defmodule MyAppWeb.ChatLive do
 use MyAppWeb, :live_view

 alias MyApp.Chat

 @impl true
 def mount(%{"room_id" => room_id}, session, socket) do
 if connected?(socket) do
 # PubSubでルームに参加
 Phoenix.PubSub.subscribe(MyApp.PubSub, "chat:#{room_id}")
 end

 user = get_user_from_session(session)
 messages = Chat.list_messages(room_id)

 {:ok,
 socket
 |> assign(:room_id, room_id)
 |> assign(:user, user)
 |> assign(:messages, messages)
 |> assign(:new_message, "")}
 end

 @impl true
 def handle_event("send_message", %{"message" => content}, socket) do
 %{room_id: room_id, user: user} = socket.assigns

 case Chat.create_message(%{
 room_id: room_id,
 user_id: user.id,
 content: content
 }) do
 {:ok, message} ->
 # 全参加者にブロードキャスト
 Phoenix.PubSub.broadcast(
 MyApp.PubSub,
 "chat:#{room_id}",
 {:new_message, message}
 )

 {:noreply, assign(socket, new_message: "")}

 {:error, _changeset} ->
 {:noreply, put_flash(socket, :error, "メッセージを送信できませんでした")}
 end
 end

 @impl true
 def handle_info({:new_message, message}, socket) do
 {:noreply, update(socket, :messages, &(&1 ++ [message]))}
 end

 @impl true
 def render(assigns) do
 ~H"""
 <div class="chat-container" id="chat" phx-hook="ScrollToBottom">
 <div class="messages">
 <%= for message <- @messages do %>
 <div class={"message #{if message.user_id == @user.id, do: "own"}"}>
 <span class="author"><%= message.user.name %>
 <p class="content"><%= message.content %></p>
 <time><%= format_time(message.inserted_at) %></time>
 </div>
 <% end %>
 </div>

 <form phx-submit="send_message" class="message-form">
 <input
 type="text"
 name="message"
 value={@new_message}
 placeholder="メッセージを入力..."
 autocomplete="off"
 />
 <button type="submit">送信</button>
 </form>
 </div>
 """
 end

 defp format_time(datetime) do
 Calendar.strftime(datetime, "%H:%M")
 end

 defp get_user_from_session(session) do
 # セッションからユーザー情報を取得
 MyApp.Accounts.get_user(session["user_id"])
 end
end

Phoenix Channels

LiveViewに加えて、Phoenix Channelsは低レベルなWebSocket通信を提供します。モバイルアプリやカスタムクライアントとの連携に最適です。

defmodule MyAppWeb.GameChannel do
 use MyAppWeb, :channel

 alias MyApp.GameServer

 @impl true
 def join("game:" <> game_id, _params, socket) do
 case GameServer.join_game(game_id, socket.assigns.user_id) do
 {:ok, game_state} ->
 send(self(), :after_join)
 {:ok, game_state, assign(socket, :game_id, game_id)}

 {:error, reason} ->
 {:error, %{reason: reason}}
 end
 end

 @impl true
 def handle_info(:after_join, socket) do
 # 他のプレイヤーに参加を通知
 broadcast!(socket, "player_joined", %{
 user_id: socket.assigns.user_id,
 timestamp: DateTime.utc_now()
 })

 {:noreply, socket}
 end

 @impl true
 def handle_in("move", %{"position" => position}, socket) do
 %{game_id: game_id, user_id: user_id} = socket.assigns

 case GameServer.make_move(game_id, user_id, position) do
 {:ok, new_state} ->
 broadcast!(socket, "game_updated", new_state)
 {:reply, :ok, socket}

 {:error, reason} ->
 {:reply, {:error, %{reason: reason}}, socket}
 end
 end

 @impl true
 def terminate(_reason, socket) do
 GameServer.leave_game(socket.assigns.game_id, socket.assigns.user_id)
 :ok
 end
end

Ectoによるデータベース操作

Ectoは、Elixirの強力なデータベースラッパーおよびクエリ言語です。型安全性と表現力を両立しています。

defmodule MyApp.Blog.Post do
 use Ecto.Schema
 import Ecto.Changeset

 schema "posts" do
 field :title, :string
 field :body, :string
 field :published, :boolean, default: false
 field :view_count, :integer, default: 0
 field :published_at, :utc_datetime

 belongs_to :author, MyApp.Accounts.User
 has_many :comments, MyApp.Blog.Comment
 many_to_many :tags, MyApp.Blog.Tag, join_through: "posts_tags"

 timestamps()
 end

 def changeset(post, attrs) do
 post
 |> cast(attrs, [:title, :body, :published, :author_id])
 |> validate_required([:title, :body, :author_id])
 |> validate_length(:title, min: 5, max: 200)
 |> validate_length(:body, min: 100)
 |> maybe_set_published_at()
 end

 defp maybe_set_published_at(changeset) do
 if get_change(changeset, :published) == true do
 put_change(changeset, :published_at, DateTime.utc_now())
 else
 changeset
 end
 end
end

高度なクエリ

defmodule MyApp.Blog do
 import Ecto.Query
 alias MyApp.Repo
 alias MyApp.Blog.Post

 # 公開済み投稿を人気順で取得
 def list_popular_posts(limit \\ 10) do
 from(p in Post,
 where: p.published == true,
 order_by: [desc: p.view_count],
 limit: ^limit,
 preload: [:author, :tags]
 )
 |> Repo.all()
 end

 # タグ別の投稿数を集計
 def count_posts_by_tag do
 from(p in Post,
 join: t in assoc(p, :tags),
 where: p.published == true,
 group_by: t.name,
 select: {t.name, count(p.id)},
 order_by: [desc: count(p.id)]
 )
 |> Repo.all()
 |> Map.new()
 end

 # 全文検索（PostgreSQL）
 def search_posts(query_string) do
 from(p in Post,
 where:
 p.published == true and
 fragment(
 "to_tsvector('english', ? || ' ' || ?) @@ plainto_tsquery('english', ?)",
 p.title,
 p.body,
 ^query_string
 ),
 order_by: [desc: p.published_at]
 )
 |> Repo.all()
 end
end

パフォーマンス特性

ベンチマーク比較

メモリ効率

# プロセスのメモリ使用量を確認
defmodule MemoryAnalyzer do
 def analyze_processes do
 Process.list()
 |> Enum.map(fn pid ->
 info = Process.info(pid, [:memory, :message_queue_len, :registered_name])
 {pid, info}
 end)
 |> Enum.sort_by(fn {_pid, info} -> -info[:memory] end)
 |> Enum.take(10)
 end

 def system_memory do
 %{
 total: :erlang.memory(:total),
 processes: :erlang.memory(:processes),
 system: :erlang.memory(:system),
 atom: :erlang.memory(:atom),
 binary: :erlang.memory(:binary),
 ets: :erlang.memory(:ets)
 }
 end
end

2025年の動向

Phoenix 1.8の新機能

2025年にリリース予定のPhoenix 1.8では、以下の新機能が期待されています。

• LiveView Streams強化: 大規模リストの効率的な更新
• コンポーネントの遅延ロード: 初期ロード時間の短縮
• 型システム統合: Elixir 1.18の型機能との連携
• Bumblebee統合: 機械学習モデルのシームレスな組み込み

Elixir 1.18の型システム

# Elixir 1.18での型アノテーション（実験的）
defmodule MyApp.Calculator do
 @spec add(number(), number()) :: number()
 def add(a, b), do: a + b

 @spec divide(number(), number()) :: {:ok, float()} | {:error, :division_by_zero}
 def divide(_a, 0), do: {:error, :division_by_zero}
 def divide(a, b), do: {:ok, a / b}
end

Nx/Bumblebeeによる機械学習

defmodule MyApp.Sentiment do
 def analyze(text) do
 {:ok, model} = Bumblebee.load_model({:hf, "distilbert-base-uncased-finetuned-sst-2-english"})
 {:ok, tokenizer} = Bumblebee.load_tokenizer({:hf, "distilbert-base-uncased"})

 serving = Bumblebee.Text.text_classification(model, tokenizer)

 Nx.Serving.run(serving, text)
 end
end

# 使用例
MyApp.Sentiment.analyze("This product is amazing!")
# => %{predictions: [%{label: "POSITIVE", score: 0.9998}]}

採用事例

Discord

Discordは、Elixirを活用して500万以上の同時接続を単一サーバーで処理しています。特にリアルタイムメッセージングとボイスチャネル管理にElixirの並行処理能力が活かされています。

Pinterest

Pinterestは、通知システムにElixirを採用し、1日あたり数十億のプッシュ通知を処理しています。

PepsiCo

PepsiCoは、サプライチェーン管理システムにElixir/Phoenixを採用し、リアルタイムな在庫追跡と需要予測を実現しています。

まとめ

Elixir/Phoenixは、2025年においても最先端のリアルタイムWeb開発プラットフォームとして進化を続けています。

主な強み

• 圧倒的な並行処理能力: BEAMの軽量プロセスによる数百万の同時接続
• 耐障害性: OTPのSupervisorパターンによる自動復旧
• 開発者体験: LiveViewによるシンプルなリアルタイムUI開発
• パフォーマンス: 低レイテンシ、高スループット
• 保守性: 関数型プログラミングによる予測可能なコード

適したユースケース

• リアルタイムチャット・メッセージング
• IoTデータ処理
• ゲームサーバー
• 金融取引システム
• ライブストリーミング

Elixir/Phoenixは、特に高い並行性と耐障害性が求められるシステムにおいて、その真価を発揮します。学習曲線は存在しますが、関数型プログラミングとOTPの概念を習得することで、堅牢でスケーラブルなシステムを効率的に構築できるようになります。

参考リソース

• Elixir公式サイト
• Phoenix Framework
• Phoenix LiveView ドキュメント
• Ecto ドキュメント
• Elixir School（日本語）

×

PR

×

PR