Elixir轻量级进程如何成就百万级并发爬虫？

defmodule ConcurrentCrawler do
  @moduledoc """
  高并发网页爬虫实现
  使用 Elixir 的并发特性和 HTTPoison 库
  """
​
  use Task
  require Logger
​
  @user_agent "Mozilla/5.0 (compatible; ConcurrentCrawler/1.0; +http://example.com/bot.html)"
  @default_concurrency 50
  @request_timeout 10_000
​
  def start_link(urls, concurrency \\ @default_concurrency) when is_list(urls) do
    Task.start_link(fn -> crawl(urls, concurrency) end)
  end
​
  @doc """
  主爬取函数
  """
  def crawl(urls, concurrency) do
    # 创建任务监督器
    {:ok, supervisor} = Task.Supervisor.start_link(max_children: concurrency)
​
    # 创建流处理管道
    urls
    |> Stream.map(&String.trim/1)
    |> Stream.filter(&valid_url?/1)
    |> Stream.uniq()
    |> Task.Supervisor.async_stream_nolink(
      supervisor,
      &process_url/1,
      max_concurrency: concurrency,
      timeout: @request_timeout * 2
    )
    |> Stream.run()
  end
​
  defp process_url(url) do
    Logger.info("Processing: #{url}")
​
    case fetch_url(url) do
      {:ok, %{status_code: 200, body: body}} ->
        # 提取数据并解析链接
        data = extract_data(body, url)
        links = extract_links(body, url)
        
        # 存储数据（这里只是打印，实际应用中可存入数据库）
        Logger.info("Extracted #{length(links)} links from #{url}")
        Logger.debug("Data: #{inspect(data, limit: 3)}")
        
        # 可以选择继续爬取新发现的链接
        # crawl(links, concurrency) # 注意：这需要额外的循环控制逻辑
        
        {:ok, url, data, links}
​
      {:ok, response} ->
        Logger.warning("Failed to fetch #{url}: Status #{response.status_code}")
        {:error, url, response.status_code}
​
      {:error, reason} ->
        Logger.error("Error fetching #{url}: #{inspect(reason)}")
        {:error, url, reason}
    end
  end
​
  defp fetch_url(url) do
    headers = [
      {"User-Agent", @user_agent},
      {"Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"}
    ]
​
    HTTPoison.get(url, headers,
      timeout: @request_timeout,
      recv_timeout: @request_timeout,
      follow_redirect: true
    )
  end
​
  defp extract_links(html, base_url) do
    case Floki.parse_document(html) do
      {:ok, document} ->
        document
        |> Floki.find("a")
        |> Floki.attribute("href")
        |> Enum.map(&absolute_url(&1, base_url))
        |> Enum.filter(&valid_url?/1)
        |> Enum.uniq()
​
      _ ->
        []
    end
  end
​
  defp extract_data(html, url) do
    # 这里可以根据需要提取特定数据
    # 例如标题、元描述、正文内容等
    
    case Floki.parse_document(html) do
      {:ok, document} ->
        title =
          document
          |> Floki.find("title")
          |> Floki.text()
          |> String.trim()
​
        description =
          document
          |> Floki.find("meta[name=\"description\"]")
          |> Floki.attribute("content")
          |> List.first()
          |> Kernel.||("")
​
        %{
          url: url,
          title: title,
          description: description,
          crawled_at: DateTime.utc_now()
        }
​
      _ ->
        %{url: url, title: "", description: "", crawled_at: DateTime.utc_now()}
    end
  end
​
  defp absolute_url(url, base) when is_binary(url) do
    case URI.parse(url) do
      %URI{scheme: scheme} when scheme in ["http", "https"] ->
        url
​
      _ ->
        base_uri = URI.parse(base)
        absolute_path = Path.expand(url, base_uri.path || "/")
        URI.merge(base, absolute_path) |> to_string()
    end
  end
​
  defp valid_url?(url) do
    case URI.parse(url) do
      %URI{scheme: scheme, host: host} when scheme in ["http", "https"] and not is_nil(host) ->
        true
      _ ->
        false
    end
  end
end
​
# 配置和启动应用
defmodule Crawler.Application do
  use Application
​
  def start(_type, _args) do
    # 配置HTTPoison
    :hackney_pool.start_pool(:crawler_pool, [
      timeout: 15_000,
      max_connections: 100
    ])
​
    children = [
      {Task.Supervisor, name: Crawler.TaskSupervisor},
      {ConcurrentCrawler, [["https://example.com/page1", "https://example.com/page2"]]}
    ]
​
    opts = [strategy: :one_for_one, name: Crawler.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

defp deps do
  [
    {:httpoison, "~> 1.8"},
    {:floki, "~> 0.32.0"},
    {:hackney, "~> 1.17"}
  ]
end

# 直接运行
urls = ["https://example.com", "https://example.org"]
ConcurrentCrawler.crawl(urls, 20) # 20是并发数
​
# 或者作为监督树的一部分启动
{:ok, pid} = ConcurrentCrawler.start_link(urls, 20)