Upstash Kafka, Redis 및 Next.js를 사용하여 실시간 채팅 앱 구축

프로젝트 설명

이 블로그 게시물에서는 사용자가 메시지 클라이언트와 채팅방을 만들 수 있는 메시징 애플리케이션을 만들겠습니다. 또한 사용자는 과거 메시지에 액세스할 수 있습니다.

프로젝트는 두 페이지로 구성됩니다. 첫 번째 페이지는 고유한 이름으로 여러 클라이언트를 생성할 수 있는 클라이언트 등록 전용 페이지입니다.

클라이언트의 사용자 이름을 클릭하면 해당 사용자와 연결된 채팅방 클라이언트로 이동됩니다.

채팅 애플리케이션의 논리는 다음과 같습니다:

사용자는 인덱스 페이지에서 각각 고유한 사용자 이름을 가진 여러 클라이언트를 만들 수 있습니다. 클라이언트의 사용자 이름을 클릭하면 고유한 경로가 있는 별도의 클라이언트가 포함된 새 탭이 열립니다.

각 클라이언트는 WebSocket 연결을 통해 메시지 서버에 연결됩니다. 클라이언트에서 새 메시지가 생성되면 해당 클라이언트와 연결된 메시지 서버로 전송됩니다.

메시지 서버는 메시지 트래픽을 처리합니다. 클라이언트가 WebSocket 연결을 통해 메시지를 보내면 서버는 해당 메시지를 Kafka Broker로 보냅니다. 각 메시지 서버는 NodeJS 스레드를 실행하여 들어오는 메시지를 처리합니다. 메시지가 소비되면 기존 WebSocket 연결을 통해 클라이언트로 전송됩니다. 클라이언트 측에서 수신 메시지를 사용하기 위해 react-use-websocket를 사용합니다. 도서관.

애플리케이션은 Upstash Redis를 활용하여 메시지 기록을 저장합니다. 메시지가 Kafka에 생성되면 Redis 데이터베이스에도 유지됩니다. 새 클라이언트를 생성하면 이전 메시지가 Upstash Redis에서 검색되어 채팅 디스플레이에 렌더링됩니다.

애플리케이션의 일반적인 개요는 다음과 같습니다.

참고: 구현에서는 데모 목적으로 단일 메시지 서버를 생성할 것이며 메시지 로드를 처리하기 위해 서버 수를 늘릴 수 있습니다.

데모

여기에서 앱 데모를 볼 수 있습니다. 현재 버전의 애플리케이션이 Fly에 배포되었습니다.

시작하기

채팅 애플리케이션을 구축하는 단계는 다음과 같습니다.

1. Upstash Redis 데이터베이스 생성
2. Upstash Kafka 클러스터 생성
3. 다음 애플리케이션 생성(프런트엔드)
4. WebSocket 메시지 서버 생성
5. Fly.io에 애플리케이션 배포

Upstash Redis 데이터베이스 생성

Upstash 콘솔로 이동하여 로그인한 다음 Redis에 로그인하세요. 탭에서 데이터베이스 만들기를 클릭하세요. 버튼을 누르세요.

이렇게 하면 Redis를 사용할 준비가 되었습니다. 자격 증명을 얻기 위해 Redis 콘솔로 돌아갑니다.

Upstash Kafka 클러스터 생성

이제 Kafka로 전환하세요. 탭을 누르고 클러스터 만들기를 클릭합니다. 버튼. 클러스터 이름을 입력하고 계속 진행합니다. 그런 다음 Kafka 주제를 생성하고 확인합니다.

다음 앱 만들기

먼저 터미널에서 애플리케이션의 루트 폴더를 만들고 탐색합니다. Next 앱과 서버는 이 폴더에 보관하겠습니다.

mkdir chat-app
cd chat-app

그런 다음 다음 앱을 만들어 보세요.

$ npx create-next-app@latest
 
✔ What is your project named? … next-chat-app
✔ Would you like to use TypeScript? … Yes
✔ Would you like to use ESLint? … Yes
✔ Would you like to use Tailwind CSS? … No
✔ Would you like to use `src/` directory? … No
✔ Would you like to use App Router? (recommended) … No
✔ Would you like to customize the default import alias? … No

자격증명 처리

.env이라는 파일을 생성하겠습니다. 자격 증명을 저장합니다. 자격 증명을 반복해서 복사하여 붙여넣을 필요 없이 이 파일에서 가져오기만 하면 됩니다.

먼저 .env를 만듭니다. 파일입니다.

그런 다음 Redis 콘솔로 이동하여 UPSTASH_REDIS_REST_URL을 복사하여 붙여넣습니다. 및 UPSTASH_REDIS_REST_TOKEN .env에 대한 자격 증명 파일입니다.

마지막으로 Kafka 콘솔로 전환하고 UPSTASH_KAFKA_REST_URL를 전송합니다. , UPSTASH_KAFKA_REST_USERNAME , UPSTASH_KAFKA_REST_PASSWORD

이제 당신의 .env 파일은 유사해 보일 것입니다

UPSTASH_REDIS_REST_URL=...
UPSTASH_REDIS_REST_TOKEN=...
 
UPSTASH_KAFKA_REST_URL=...
UPSTASH_KAFKA_REST_USERNAME=...
UPSTASH_KAFKA_REST_PASSWORD=...

이제 자격 증명을 구성했으므로 신청을 진행할 수 있습니다.

클라이언트 등록 페이지

인덱스 페이지에는 클라이언트 등록/생성 작업이 포함됩니다. 사용자 이름을 제출하면 새 클라이언트가 생성되어 현재 클라이언트 아래에 나열됩니다. 테이블.

import { useState } from "react";
import Link from "next/link";
 
import { Redis } from "@upstash/redis";
 
import styles from "@/styles/Home.module.css";
 
export default function Home() {
 const [usernameInput, setUsernameInput] = useState<string>("");
 const [usernameList, setUsernameList] = useState<string[]>(Array<string>);
 
 const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>): void => {
 const inputValue: string = e.target.value;
 setUsernameInput(inputValue);
 };
 
 const addUsernameClient = (e: React.FormEvent<HTMLFormElement>): void => {
 e.preventDefault();
 setUsernameList([...usernameList, usernameInput]);
 setUsernameInput("");
 };
 return (
 <div className={styles.container}>
 <div className={styles.welcomeSection}>
 <h1>Welcome to the demo message app!</h1>
 <p>
 This application uses Upstash Kafka for message passing, and Upstash
 Redis for state management.
 <br />
 <br />
 To get started, create several clients by typing in unique usernames to
 the input section below and submitting.
 <br />
 <br />
 The usernames will be added to the list of current clients. Click on a
 username to open a new tab with that client&apos;s message display.
 <br />
 <br />
 You can have multiple sessions open at once.
 </p>
 </div>
 <form className={styles.formSection} onSubmit={addUsernameClient}>
 <input
 type="text"
 className={styles.formInput}
 value={usernameInput}
 onChange={handleInputChange}
 ></input>
 
 <button className={styles.formSubmit} type="submit">
 Create the client!
 </button>
 </form>
 <div className={styles.clientListSection}>
 <p className={styles.clientListHeader}>Current Clients</p>
 <div className={styles.clientList}>
 {usernameList.map((username, i) => {
 return (
 <Link
 href={`/user/${username}`}
 key={`${i}`}
 className={styles.userClient}
 target={"_blank"}
 >
 <p>{username}</p>
 </Link>
 );
 })}
 </div>
 </div>
 </div>
 );
}

앱을 다시 로드할 때마다 채팅 기록을 재설정하려면 다음 기능을 사용하세요:

export async function getServerSideProps() {
 const redis = new Redis({
 url: process.env.UPSTASH_REDIS_REST_URL,
 token: process.env.UPSTASH_REDIS_REST_TOKEN,
 });
 
 await redis.del("messagesList");
 
 return {
 props: {},
 };
}

이로써 인덱스 페이지를 실행할 준비가 되었습니다. npm run dev 실행 next-chat-app의 명령 폴더를 열면 색인 페이지가 활성화되는 것을 볼 수 있습니다.

메시지 클라이언트 페이지

클라이언트에 대한 동적 라우팅을 구현하기 위해 /pages/user/[username].tsx라는 폴더를 생성합니다. 이 폴더 구조를 사용하면 사용자 이름을 기반으로 각 개별 클라이언트에 대한 동적 경로를 생성할 수 있습니다.

다음은 클라이언트의 주요 구성 요소입니다. 이 구성 요소는 메시지 목록, 사용자 이름 등에 대한 상태를 보유합니다. useWebSocket 후크를 사용하여 WebSocket에서 메시지, 연결 및 연결 끊김 이벤트를 생성합니다. 메시지 이벤트가 발생하면 메시지가 메시지 목록에 추가되고 MessageDisplay 구성 요소가 다시 렌더링됩니다.

import { useState } from "react";
import { useRouter } from "next/router";
 
import { Redis } from "@upstash/redis";
import useWebSocket from "react-use-websocket";
 
import styles from "@/styles/Home.module.css";
 
type Message = {
 id: number;
 sender: string;
 text: string;
};
 
export default function MessageApp(props: { messagesData: Message[] }) {
 const { messagesData } = props;
 const { username } = useRouter().query;
 const [inputText, setInputText] = useState<string>("");
 const [messageList, setMessageList] = useState<Message[]>(messagesData);
 const [messageCounter, setMessageCounter] = useState<number>(0);
 
 const handleMessage = function (message: Message) {
 const nextMessages = [...messageList, message];
 setMessageList(nextMessages);
 };
 
 // handling WebSocket events
 const { sendMessage } = useWebSocket("ws://localhost:8080", {
 share: true,
 filter: () => false,
 onOpen: () => {
 console.log("WebSocket connection!");
 return "connection";
 },
 
 onMessage: (message) => {
 const data = JSON.parse(message.data);
 const { sender, text }: { sender: string; text: string } = data;
 const messageData: Message = {
 id: messageCounter,
 sender: sender,
 text: text,
 };
 setMessageCounter(messageCounter + 1);
 handleMessage(messageData);
 return message;
 },
 
 onClose: () => {
 console.log("WebSocket disconnected!");
 return "disconnected";
 },
 });
 
 function handleSendMessage(messageText: string) {
 const messageData = {
 sender: username,
 text: messageText,
 };
 
 sendMessage(JSON.stringify(messageData));
 }
 
 return (
 <div className={styles.Container}>
 <MessageDisplay messages={messageList} />
 <MessageInput
 inputText={inputText}
 setInputText={setInputText}
 handleSendMessage={handleSendMessage}
 />
 </div>
 );
}

MessageDisplay 및 MessageInput 구성 요소는 다음과 같습니다.

const MessageDisplay = function (props: { messages: Message[] }) {
 const { messages } = props;
 
 return (
 <div className={styles.messageContainer}>
 {messages.map((message) => (
 <MessageBubble
 key={message.id}
 sender={message.sender}
 text={message.text}
 />
 ))}
 </div>
 );
};
 
const MessageInput = (props: {
 inputText: string;
 setInputText: (msg: string) => void;
 handleSendMessage: (msg: string) => void;
}) => {
 const { inputText, setInputText, handleSendMessage } = props;
 
 const handleInputChange = (
 e: React.ChangeEvent<HTMLInputElement>
 ): void => {
 const inputValue: string = e.target.value;
 setInputText(inputValue);
 };
 
 const handleSubmit = (e: React.FormEvent<HTMLFormElement>): void => {
 e.preventDefault();
 handleSendMessage(inputText);
 if (inputText.trim() !== "") {
 setInputText(" ");
 }
 };
 
 return (
 <form className={styles.inputSection} onSubmit={handleSubmit}>
 <input
 className={styles.inputText}
 type="text"
 value={inputText}
 onChange={handleInputChange}
 ></input>
 <button className={styles.inputSendButton} type="submit">
 Send
 </button>
 </form>
 );
};
 
const MessageBubble = (props: {
 sender: string;
 text: string;
 key: number;
}) => {
 const { sender, text } = props;
 
 const { username } = useRouter().query;
 
 const isSender = sender === username;
 const senderClass = isSender ? "sender" : "receiver";
 return (
 <div className={`${styles["messageBubble"]} ${styles[senderClass]}`}>
 <div className={styles.messageSender}>
 {isSender ? "You" : sender}
 </div>
 <div className={styles.messageText}>{text}</div>
 </div>
 );
};

클라이언트에게 채팅 기록을 제공하기 위해 getServerSideProps()을 사용합니다. 기능입니다.

export async function getServerSideProps() {
 const redis = new Redis({
 url: process.env.UPSTASH_REDIS_REST_URL,
 token: process.env.UPSTASH_REDIS_REST_TOKEN,
 });
 
 const messagesData = (await redis.lrange("messagesList", 0, -1)).reverse();
 
 return {
 props: {
 messagesData,
 },
 };
}

이제 Next.js 앱이 작동 중입니다. 페이지를 새로 고치고 클라이언트를 생성한 후 그 중 하나를 탐색하세요. 클라이언트 페이지가 표시됩니다. 하지만 여전히 메시지 흐름을 처리하려면 메시지 서버가 필요합니다.

메시지 서버 생성

서버의 구조는 다소 간단합니다. 우리는 Node.js, ws 라이브러리, Upstash Kafka를 사용하여 작동하도록 할 것입니다. 먼저 server을 만듭니다. chat-app folder 내부 폴더 .

mkdir server
cd server

server 내부 폴더에 요구사항을 설치하고 파일을 구성하겠습니다.

npm install typescript ws tsc @upstash/kafka @types/ws
tsc --init

그런 다음 /server/message_server.ts 내에 WebSocket, Kafka Producer 및 Kafka Consumer 클라이언트를 생성하겠습니다. 파일:

import * as http from "http";
 
import { Kafka } from "@upstash/kafka";
import { Redis } from "@upstash/redis";
import { WebSocket } from "ws";
 
const server = http.createServer();
const wss = new WebSocket.Server({ server });
 
server.listen(8080, () => {
 console.log("Server is running on port 8080");
});
 
const kafka = new Kafka({
 url: process.env.UPSTASH_KAFKA_REST_URL,
 username: process.env.UPSTASH_KAFKA_REST_USERNAME,
 password: process.env.UPSTASH_KAFKA_REST_PASSWORD,
});
 
const redis = new Redis({
 url: process.env.UPSTASH_REDIS_REST_URL,
 token: process.env.UPSTASH_REDIS_REST_TOKEN,
});
 
const consumer = kafka.consumer();
 
const producer = kafka.producer();
 
const clients = new Set<WebSocket>();

WebSocket과 상호작용하기 위해 connection을 생성합니다. 및 message 이벤트.

wss.on("connection", async (connection, req) => {
 clients.add(connection);
 console.log(`New client connected!`);
 
 connection.on("message", async (message) => {
 const jsonMessage = message.toString();
 
 console.log("Received message:", JSON.parse(jsonMessage));
 
 producer.produce("chat", jsonMessage);
 });
 
 connection.on("close", () => {
 console.log(`Client disconnected:`);
 clients.delete(connection);
 });
});

마지막으로 미리 정의된 간격으로 메시지를 소비하는 스레드를 생성하고 실행하겠습니다.

async function run() {
 while (true) {
 const messages = await consumer.consume({
 consumerGroupId: "group_1",
 instanceId: "instance_1",
 topics: ["chat"],
 autoOffsetReset: "earliest",
 });
 
 if (messages.length != 0) {
 for (let i = 0; i < messages.length; i++) {
 await redis.lpush("messagesList", messages[i].value);
 console.log(`Message sending: ${messages[i].value}`);
 
 clients.forEach((connection: WebSocket) => {
 connection.send(messages[i].value);
 });
 }
 }
 
 console.log("Run!");
 
 await new Promise((resolve) => setTimeout(resolve, 1000));
 }
}

모든 것이 준비되었습니다. 우리 앱은 지금 당장 매력적으로 작동할 것입니다. 로컬에서 메시지 서버를 실행하고 클라이언트 페이지를 새로 고치면 클라이언트 간에 전송되는 메시지를 확인할 수 있습니다. 아래 명령은 TS 파일을 트랜스컴파일하고 localhost:8000에서 서버를 실행합니다.

tsc message_server.ts
node message_server.js

배포

배포에는 Fly.io를 사용하겠습니다. 아직 계정이 없다면 시작하기 전에 계정을 만드세요.

메시지 서버 배포

server으로 이동 폴더를 선택하고 flyctl를 설치하세요. CLI 도구 및 셸을 통한 승인

npm install flyctl
flyctl auth login

구성 파일을 생성하려면 flyctl init을 실행하세요. . 그러면 fly.toml이 생성됩니다. . fly.toml으로 이동 WebSocket 연결 구성을 위해 다음 줄을 삽입하세요:

[[services]]
 internal_port = 8080
 protocol = "tcp"
 
 [services.concurrency]
 hard_limit = 25
 soft_limit = 20
 
 [[services.ports]]
 handlers = ["http"]
 port = "80"
 
 [[services.ports]]
 handlers = ["tls", "http"]
 port = "443"
 
 [[services.tcp_checks]]
 interval = 10000
 timeout = 2000

이제 서버의 마지막 단계입니다. flyctl deploy 실행 , 이제 갈 준비가 되었습니다! 배포 프로세스가 완료되면 flyctl은 서버에 대한 끝점을 제공합니다. 해당 엔드포인트를 복사하세요. 우리의 경우 엔드포인트는 message-server.fly.dev입니다. .

다음 애플리케이션 배포

Next.js 애플리케이션을 배포하기 전에 메시지 서버의 배포 끝점을 포함해야 합니다. pages/user/[username].tsx에서 WebSocket URL을 바꾸십시오. ws://localhost:8080의 파일 flyctl에서 엔드포인트로 , wss://와 결합 접두사. 우리의 경우에는 wss://message-server.fly.dev입니다. .

그런 다음 next-chat-app에서 폴더에서 server와 동일한 명령을 실행합니다. . 이번에는 fly.toml을 편집할 필요가 없습니다. 해당 단계 없이 진행할 수 있습니다.

flyctl init
flyctl deploy

우리는 끝났습니다! flyctl open를 실행하면 명령을 실행하면 배포된 프로젝트로 이동됩니다.

결론 및 제안

따라와주셔서 감사합니다!

여기에서 프로젝트의 Github 저장소를 찾을 수 있습니다.

프로젝트를 계속 진행하고 싶다면 다음과 같은 몇 가지 제안 사항을 따르세요.

• 현재 페이지가 다시 로드될 때마다 Upstash Redis에 저장된 모든 메시지가 플러시됩니다. 이 동작은 pages/index.tsx의 코드로 제어됩니다. 파일, 특히 getServerSideProps 내의 기능. 그러나 사용자가 페이지를 다시 로드하기로 결정하면 채팅방에 참여한 모든 참가자의 채팅 기록이 삭제되는 중요한 문제가 발생합니다.
  이 문제를 해결하기 위해 권장되는 솔루션은 메시지가 전송될 때마다 채팅 기록에 대한 TTL 확장을 구현하는 것입니다. 이러한 개선을 통해 페이지를 다시 로드한 후에도 채팅 기록에 계속 액세스하고 보존할 수 있습니다.

• 여러 대화방 기능을 구현할 수 있습니다. 이를 달성하기 위해 각 채팅방에 대해 고유한 이름을 가진 여러 Kafka 주제를 만들 수 있습니다. 또 다른 방법은 올바른 데이터 구조를 사용하여 메시지 서버 자체에서 이를 처리하는 것입니다.

• 또한 여러 메시지 서버와 로드 밸런서를 구현하여 최상의 시스템 설계 방식을 적용할 수도 있습니다.

질문이 있으시면 fahreddin@upstash.com으로 연락주세요