ওয়াটারমিল প্রকাশক কম্পোনেন্ট (মেসেজ লেনদেন)

লেনদেনে বার্তা প্রকাশ করার গুরুত্ব

ইভেন্ট-ড্রিভেন অ্যাপ্লিকেশন সঙ্গে কাজ করার সময়, বোধগম্য হতে পারে যে আপনার অ্যাপ্লিকেশন অবস্থা অবস্থিতি সংরক্ষণ করতে এবং অন্য অংশগুলিকে নোটিফাই করতে বার্তা প্রকাশ করতে প্রয়োজন হতে পারে। একটি আদর্শ পরিস্থিতিতে, আপনি একটি এপ্লিকেশন স্থিতি সংরক্ষণ করা এবং একটি বার্তা প্রকাশ করতে একটি সিঙ্গল লেনদেনের মধ্যে চাইতে পারেন, কারণ এটা করা না বিপদের ডেটা সঠিকতা সমস্যায় পার্থক্য হতে পারে। একটি ট্রান্সয়াকশনে ডেটা স্টোরেজ সহ ইভেন্ট প্রকাশ সময়ে সঙ্গতি প্রভুত করতে, আপনারকে একই ডাটাবেসে বার্তা প্রকাশ করার জন্য বার্তা প্রকারের ব্যবহার করতে হয়, বা একটি দুটি ফেজ কমিট (2PC) ব্যবহার করতে হয়। যদি আপনি বার্তা ব্রোকারকে ডাটাবেসের জন্য পরিবর্তন করতে না চান এবং চাকরি পুনর্জন্ম দেওয়া না চান, তবে আপনি ওয়াটারিলের ফরওয়ার্ডার কম্পোনেন্ট-টি ব্যবহার করে আপনার কাজটি সহজ করতে পারেন!

ফরওয়ার্ডার কম্পোনেন্ট

আপনি ফরওয়ার্ডারকে একটি ব্যাকগ্রাউন্ড ডেমন ভাবতে পারেন যা ডেটাবেসে প্রকাশিত হওয়া বার্তাগুলির জন্য অপেক্ষা করে এবং নিশ্চিত করে যে তারা পরে কখনোই বার্তা ব্রোকারের কাছে পৌঁছাতে পারে।

ফরওয়ার্ডারকে সাধারণ এবং ট্রান্সপ্যারেন্ট করার জন্য, এটি একটি উপস্থিত ডেটাবেসে একটি প্রতিষ্ঠিত টপিকে শোনা বা এনক্রিপ্ট করা Forwarder publisher-এর ব্যবহার করে ক্ষণস্থলে বার্তা প্রেরণ করে। ফরওয়ার্ডার তাদের আনপ্যাক করে এবং তাদেরকে নির্দিষ্ট লক্ষ্যযুক্ত বার্তা ব্রোকারের উপর পাঠায়।

উদাহরণ

আমরা নিম্নলিখিত উদাহরণটি মনে করি: একটি কমান্ড আছে যা লটারি আটক নিয়ন্ত্রণ করে। এটির কাজ করতে হবে নিবন্ধিত ব্যবহারকারীর মধ্যে একজন ভাগ্যবান বিজয়ী হিসাবে স্বয়ংক্রিয়ভাবে নির্বাচন করা। এটি এটা করার সময়, এটা আরও তার নির্ধারিত করতে হবে একটি ডাটাবেস এন্ট্রি স্টোর করে যা অনন্য লটারি আইডি সাথে নির্বাচিত ব্যবহারকারীর আইডির সাথে সংযুক্ত হয়। আরোপরিবর্তনে, একটিভ ইভেন্ট-ড্রিভেন সিস্টেম হিসাবে, এটা ছাড়াও প্রকাশ করা দরকার LotteryConcluded ইভেন্ট, যাতে অন্যান্য কোম্পোনেন্টগুলি যথাযথ প্রতিক্রিয়া দিতে পারে। নিশ্চিত হওয়ার - একটি কোম্পোনেন্ট বিজয়ীকে পুরস্কার প্রেরণের জন্য দায়িত্বপ্রাপ্ত হবে। এটি প্রাপ্ত জয়ীর সাথে LotteryConcluded ইভেন্টটি প্রাপ্ত করবে এবং নির্ধারণ করবে ডেটাবেস এনট্রির সাথে এম্বেডেড লটারি আইডি ব্যবহার করে।

আমাদের সমর্থনে, ডেটাবেসটি মাইএসকিউএল এবং বার্তা ব্রোকারটি গুগল পাব/সাব, তবে এটিও অন্যান্য দুটি প্রযুক্তি হতে পারে।

এমনকি কমান্ডটি পারিভাষিক করার সময়, বিভিন্ন পদক্ষেপ নিতে পারেন। এইচডিএফার পরিযাল্পনা এবং তাদের অভিযোগ করার সময়ে তাকে মনে রাখা হবে।

প্রথমে ইভেন্ট প্রকাশ করুন, তারপরে ডেটা সংরক্ষণ করুন

এই পদ্ধতিতে, কমান্ডটি প্রথমে একটি ইভেন্ট প্রকাশ করে তারপরে ডেটা সংরক্ষণ করে। যদিও এই পদ্ধতি অধ্যায় এই অনেক সময় সঠিকভাবে কাজ করতে পারে, তবে আমরা প্রয়োজনীয় সমস্যা চেষ্টা করতে চাই।

কমান্ডটি তিনটি মৌলিক অপারেশন সম্পাদন করতে হবে:

1. একটি বিজয়ী হিসেবে একটি যোগাযোগের জন্য একটি একক ব্যবহারকারী 'A' নির্বাচন করুন।
2. ঘড়ি শেষ হয়েছে বোঝাতে একটি লটারি সমাপ্ত ইভেন্ট প্রকাশ করুন।
3. ডেটাবেসে বিজয়ী ব্যবহারকারী 'A' দ্বারা লটারি 'B' জিতেছে এটি সংরক্ষণ করুন।

প্রতিটি ধাপ ব্যাপক দুর্বল, যা আমাদের কমান্ড ফ্লো বিদ্যমান করতে পারে বিধ্বস্ত করতে পারে। যদি প্রথম ধাপ ব্যর্থ হয়, পরিণামগুলি খুব ভয়ংকর থাকতে পারে - আমাদের কেবল একটি ত্রুটি ফিরিয়ে পাঠাতে হবে এবং সম্পূর্ণ কমান্ডটি ব্যর্থ হিসেবে গণ্য করতে হবে। কোনও ডেটা সংরক্ষণ করা হবে না, এবং কোনও ইভেন্ট প্রকাশ করা হবে না। আমরা কেবল কমান্ডটি পুনরায় চালাতে পারি।

যদি দ্বিতীয় পদক্ষেপ ব্যর্থ হয়, আমরা এখনও ইভেন্ট প্রকাশ বা ডেটাবেসে ডেটা সংরক্ষণ করে নাই। আমরা কমান্ডটি পুনরায় চালাবো এবং আবার চেষ্টা করতে পারি।

সবচেয়ে দর্শনীয় সংলাপ টীম তৃতীয় পদক্ষেপ ব্যর্থ হলে কি হয়, এটি। দ্বিতীয় পদক্ষেপের পরে, আমরা ইভেন্ট প্রকাশ করেছি এবং চূড়ান্ত অবজেক্ট হয়নি। ডেটাবেসে কোনও ডেটা সংরক্ষণ হবে না। অন্যান্য উপাদানরা পাবে যে লটারি সমাপ্ত হয়েছে, তবে ইভেন্টে প্রেরিত লটারি আইডির সাথে কোনও বিজয়ী যুক্ত থাকবে না। তারা বিজয়ী যাচাই করতে পারবে না, তাদের অপারেশনগুলিও ব্যার্থ হিসাবে গণ্য করা হবে।

এই অবস্থা থেকে আমরা এখনও বের হতে পারি, তবে এটা কিছু ম্যানুয়াল অপারেশন প্রয়োজন করতে পারে, যেমন ইভেন্ট থেকে লটারি আইডি ব্যবহার করে কমান্ডটি আবার চালানো।

সম্পূর্ণ উৎস কোড: github.com/ThreeDotsLabs/watermill/_examples/real-world-examples/transactional-events-forwarder/main.go

// ...
// 1. প্রথমে Google Cloud Pub/Sub-এ ইভেন্ট প্রকাশ করুন, তারপরে MySQL-এ ডেটা সংরক্ষণ করুন।
func publishEventAndPersistData(lotteryID int, pickedUser string, logger watermill.LoggerAdapter) error {
	publisher, err := googlecloud.NewPublisher(
		googlecloud.PublisherConfig{
			ProjectID: projectID,
		},
		logger,
	)
	if err != nil {
		return err
	}
	event := LotteryConcludedEvent{LotteryID: lotteryID}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}
	err = publisher.Publish(googleCloudEventTopic, message.NewMessage(watermill.NewULID(), payload))
	if err != nil {
		return err
	}

	// যদি কোনও ত্রুটি থাকে, ইভেন্ট প্রেরিত করা হয়েছে, তবে এখনও ডেটা সংরক্ষণ করা হয়নি।
	if err = simulateError(); err != nil {
		logger.Error("ডেটা সংরক্ষণ অসম্পূর্ণ", err, nil)
		return err
	}

	_, err = db.Exec(`INSERT INTO lotteries (lottery_id, winner) VALUES(?, ?)`, lotteryID, pickedUser)
	if err != nil {
		return err
	}
// ...

প্রথমে ডেটা সংরক্ষণ করুন, তারপরে ইভেন্ট প্রকাশ করুন

দ্বিতীয় পদ্ধতিতে, আমরা প্রথম অ্যাড্রেস পদ্ধতির অপুরণ হাতার সমস্যাগুলির সমাধান করার চেষ্টা করব। ডেটাবেসে স্থিতি সঠিকভাবে সংরক্ষিত না হওয়া এবং ইভেন্ট প্রকাশ না হওয়া সময় আউটার কম্পোনেন্টগুলিতে ব্যাপ্ত হয় না এর জন্য, আমরা ক্রিয়াবলি পরিবর্তন করব:

1. বিবিধ লটারি B-র বিজয়ী হিসেবে ব্যবহারকারী A পুনঃনির্ধারিত করুন।
2. তথ্য সংরক্ষণ করুন যে লটারি B ব্যবহারকারী A দ্বারা জিতেছে ডেটাবেসে।
3. LotteryConcluded ইভেন্ট প্রকাশ করুন, যা লটারি B শেষ হয়েছে সেই সূচিত করে।

প্রথম পদ্ধতির মত যদি প্রথম দুটি ক্রিয়া ব্যর্থ হয়, তবে আমাদের কোনও ফলাফল থাকবে না। তৃতীয় ক্রিয়ার ব্যর্থতার মামলায়, আমাদের ডেটা ডেটাবেসে সংরক্ষণ করা হবে, কিন্তু কোনও ইভেন্ট প্রকাশ করা হবে না। এই মামলায়, আমরা লটারির বাইরের কম্পোনেন্টগুলিতে বিফলতা সংকেত দেওয়া হবে না। তাছাড়া, আশাভাবে সিস্টেমের আনুমানিক আচরণ বিবেক করে, আমাদের বিজয়ী পুরস্কার পাবে না কারণ এই অপারেশনের জন্য কোনও ইভেন্ট পাঠানো হয়নি।

এই সমস্যাটি হাতে করার মাধ্যমও, অর্থাৎ, ইভেন্ট আবিষ্কার করার মাধ্যমে সমাধান করা যেতে পারে। কিন্তু আমরা আরও ভালো করতে পারি।

সম্পূর্ণ উৎস কোড: github.com/ThreeDotsLabs/watermill/_examples/real-world-examples/transactional-events-forwarder/main.go

// ...
// 2. প্রথমে ডেটা MySQL তে সংরক্ষণ করুন, তারপরে সরাসরি Google Cloud Pub/Sub এ ইভেন্ট প্রকাশ করুন।
func persistDataAndPublishEvent(lotteryID int, pickedUser string, logger watermill.LoggerAdapter) error {
	_, err := db.Exec(`INSERT INTO lotteries (lottery_id, winner) VALUES(?, ?)`, lotteryID, pickedUser)
	if err != nil {
		return err
	}

	var publisher message.Publisher
	publisher, err = googlecloud.NewPublisher(
		googlecloud.PublisherConfig{
			ProjectID: projectID,
		},
		logger,
	)
	if err != nil {
		return err
	}

	event := LotteryConcludedEvent{LotteryID: lotteryID}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}

// এই যদি ব্যর্থ হয়, তাহলে আমাদের ডেটা ইতিমধ্যে সংরক্ষিত, কিন্তু কোনও ইভেন্ট প্রকাশ করা হয়নি।
	if err = simulateError(); err != nil {
		logger.Error("ইভেন্ট প্রকাশ করতে ব্যর্থ", err, nil)
		return err
	}

	err = publisher.Publish(googleCloudEventTopic, message.NewMessage(watermill.NewULID(), payload))
	if err != nil {
		return err
	}
// ...

লেনঘা ডেটা এবং ঘটনা প্রকাশ রাখা একটি লেনঘায়নে

মনে করেন, আমাদের কমান্ড সময়ের সাথে দ্বিতীয় এবং তৃতীয় কাজ পালন করতে পারে। এগুলি পরামাণুতা প্রাপ্ত হবে, অর্থাৎ যদি একটি কাজ ব্যর্থ হয়, অন্যটি সফল হতে পারবে না। এটি সব ডাটাবেসেইতে ইমপ্লিমেন্ট করা লেনঘা ব্যবহার করে এটি সম্ভব। আমাদের উদাহরণে ব্যবহৃত MySQL একটি তারা।

ডাটা স্টোর এবং ঘটনা ট্রাঞ্জ্যাকশানে রাখতে, আমাদেরকে মানসপতি ডেটাবেসে বার্তাগুলি প্রকাশ করতে পারা দরকার। যেহেতু আমরা চাই না এই পূর্ণ সিস্টেম পর্যন্ত MySQL দ্বারা সমর্থিত বার্তা ব্রোকারকে পরিবর্তন করতে, সে কাজ করার অন্য উপায় আমাদের অবশ্যই খোঁজা লাগবে।

সুন্দর খবর হল: Watermill সমস্ত প্রয়োজনীয় সরঞ্জাম সরবরাহ করে! আপনি যদি MySQL, PostgreSQL (বা অন্য কোনও SQL ডাটাবেস), Firestore বা Bolt এরকম ডাটাবেস ব্যবহার করেন, তবে আপনি তাদের প্রতি বার্তা প্রকাশ করতে পারবেন। Forwarder মোডিউল আপনাকে ডাটাবেসে যেগুলি আপনি প্রকাশ করেছেন সিলেক্ট করতে এবং তাদেরকে আপনার বার্তা ব্রোকারে ফরোয়ার্ড করতে সাহায্য করবে।

আপনাকে নিশ্চিত করতে হবে যে:

1. আপনার কমান্ডটি একটি লেনঘা লেন্দের সংদর্ভে কাজ করে (উদাহরণস্বরূপ SQL, Firestore, Bolt)।
2. Forwarder মোডিউল চালু আছে, ডাটাবেস সাবস্ক্রাইবার এবং বার্তা ব্রোকার পাবলিশার মাধ্যমে।

এই ক্ষেত্রে, কমান্ডটি দেখতে নিম্নরূপ:

Complete source code: github.com/ThreeDotsLabs/watermill/_examples/real-world-examples/transactional-events-forwarder/main.go

Forwarder মোডিউলকে পসার করার জন্য আপনাকে পিছিয়ে কাজ করতে হবে এবং মেসেজগুলি মাইক্রোসফট পাব / সাব এ ফরোয়ার্ড করতে, আপনাকে এটি নিম্নের মত সেট করতে হবে:

Complete source code: github.com/ThreeDotsLabs/watermill/_examples/real-world-examples/transactional-events-forwarder/main.go

// ...
// একটি লেন্ধনে ডেটা পারিবারিকভাবে সংরক্ষণ করুন এবং গুগল ক্লাউড পাব / সাবে ইভেন্ট প্রকাশ করুন।
func persistDataAndPublishEventInTransaction(lotteryID int, pickedUser string, logger watermill.LoggerAdapter) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	defer func() {
		if err == nil {
			tx.Commit()
		} else {
			logger.Info("Rolling back the transaction due to an error", watermill.LogFields{"error": err.Error()})
			// ত্রুটির কারণে, MySQL লেন্ধন রোলব্যাকের কারণে, আমরা নিশ্চিত করতে পারি যে নিম্নলিখিত অবাঞ্ছিত সেনারিও ঘটনা ধারণ না করে:
			// - ইভেন্ট প্রকাশ করা হয়েছে, কিন্তু ডেটা সংরক্ষিত হয়নি,
			// - ডেটা সংরক্ষিত, কিন্তু ইভেন্ট প্রকাশ হয়নি।
			tx.Rollback()
		}
	}()

	_, err = tx.Exec(`INSERT INTO lotteries (lottery_id, winner) VALUES(?, ?)`, lotteryID, pickedUser)
	if err != nil {
		return err
	}

	var publisher message.Publisher
	publisher, err = sql.NewPublisher(
		tx,
		sql.PublisherConfig{
			SchemaAdapter: sql.DefaultMySQLSchema{},
		},
		logger,
	)
	if err != nil {
		return err
	}

	// পাবলিশারের সাথে একটি কভাপড দিয়ে যোগ করুন যা ফরওয়ার্ডার মডিউল বুঝতে পারে।
	publisher = forwarder.NewPublisher(publisher, forwarder.PublisherConfig{
		ForwarderTopic: forwarderSQLTopic,
	})

	// লটারি শেষবেলা ইভেন্টের ঘোষণা করুন। দয়া করে মনতে রাখুন, এখানে আমরা MySQL পাবলিশারটিকে সাজানো জানার জন্য একটি গুগল ক্লাউড টপিকে প্রকাশ করছি।
	event := LotteryConcludedEvent{LotteryID: lotteryID}
	payload, err := json.Marshal(event)
	if err != nil {
		return err
	}

	err = publisher.Publish(googleCloudEventTopic, message.NewMessage(watermill.NewULID(), payload))
	if err != nil {
		return err
	}

	return nil
// ...

যদি আপনি এই উদাহরণে আরও জানতে চান, তবে এর অনুসন্ধান করুন [এখানে](https://github.com/ThreeDotsLabs/watermill/tree/master/_examples/real-world-examples/transactional-events-forwarder/main.go)।