How To Upload And Download File From Java To Amqp

Work queue with Go and RabbitMQ

We want to build a work queue arrangement. We want to maintain a queue of tasks / work where we would push new work. Workers will actively monitor the queue and practise these work every bit they come. A work queue organization is platonic for background jobs which tin have time, take longer than your average http request. A mutual example I apply ofttimes to depict work queues — if your app handles user uploaded photos, creates several versions (thumbnails, different sizes) and shares them on different social media, how would y'all handle these? Resizing the photos and uploading them on other sites will definitely take time. Practise y'all want to do this inside your http handler? You lot should not. Instead, when a photo is uploaded, shop the photo somewhere convenient and laissez passer the details to background workers which will then process the photo and upload wherever needed.

We need a messaging system

To accomplish what we want to practice, we need a arrangement that can work every bit a queue of messages. We want to have a distributed piece of work queue system, where the workers and the work producers would alive on unlike servers. At that place tin be many source of work and many workers processing those. To build such a distributed arrangement, we demand a centralized bulletin queue / message broker. Nosotros need a system where we can pass messages and this system volition deliver these messages to our workers.

At that place are several tools that can help us with information technology — Redis, Kafka, RabbitMQ, ZeroMQ, IronMQ, AWS SQS — at that place are enough of choices. Redis is very popular and I personally use it a lot with Celery / Bull. Only RabbitMQ is a amend choice in some aspects. In our case, we're going to use RabbitMQ with Become to create a very simple work queue system.

Understanding RabbitMQ concepts

There are several concepts we should go on in heed while working with RabbitMQ.

Producers & Consumers: A producer is someone who creates new messages / work. Consumers are who swallow them. In our example, when a file upload happens, the http handler produces a bulletin for our workers to consume. The http handler / web app is the producer, the background workers are the consumers.

Exchange and Queues: From the name, you lot tin can approximate they are somehow related to message handling. Exchanges receive messages from producers and deliver them to queues. Consumers consume letters from the queues. RabbitMQ comes with very powerful message routing features. We can, in many ways, customize the way messages are delivered to the different queues.

A command line example

In this mail service, we really don't desire to go into building a total fledged web app with file uploads. We want to continue it brusque. So nosotros volition do things on the command line. Nosotros will build a command line publisher tool which will publish/produce messages. And a consumer which will consume letters. We will then run multiple instances of the consumers in parallel to show how we can calibration this organisation by calculation more than workers.

We are going to build a very sophisticated (?!?) calculator that can take two numbers and output their sum on the standard output. And we need to make information technology web scale, and so we demand to use Go and RabbitMQ.

Initial setup

Before we tin start building out 1000000 dollar reckoner, we need to take RabbitMQ installed somewhere we can connect to. For the states developers, what'south a meliorate identify than localhost? Let's install RabbitMQ on our local machine and have it running.

The RabbitMQ installation might vary from platform to platform. On a MacBook, I installed using Homebrew. On a linux distro, it's probably available from the packet manager. For Windows, there should be installable packages.

Once we accept RabbitMQ installed, nosotros need to have the Go AMQP packet installed on our system. I am using Become Modules to install it. You lot may go get it, or use a dependency management arrangement.

          go get github.com/streadway/amqp

Building a consumer

We created a consumer directory inside which, we will build our consumer app. The consumer app needs to connect to RabbitMQ, declare a queue it wants to mind to and then starting time consuming letters.

Before we begin, we're going to build an error treatment function. This should help the states through the tedious fault treatment scenario of Go.

If an error is not nil, print a message, the error details and quit. That's what the above part does.

Now let's kickoff setting up a connectedness to RabbitMQ.

We're going to endeavor to connect to RabbitMQ and quit if it fails. The connection url is stored in a top level shared.go file. The value is set to: "amqp://invitee:guest@localhost:5672/".

If the connection succeeds, nosotros need to establish a channel. Do not misfile it with Go's channel. RabbitMQ has it's own concept of channels. A connection is a TCP connection from the client to server. A connection is not cheap to create. A channel serves as the advice protocol over the connection. Channels are quite cheap. We should aim to limit connections to a minimum number while establishing as many channels as nosotros need, on top of these connections.

We can now get-go talking to RabbitMQ. Nosotros need tell the server virtually the queue we're interested in.

Next we create a queue named add. If you're curious what the function arguments are, take a look at the docs here.

RabbitMQ starts delivering letters to consumers in round robin mode. So it equally distributes work among all workers. If some piece of work accept way longer and some terminate very first, one worker will accept a lot of accumulated tasks, another one will not break a sweat at all. 1 worker will always be busy, one will be e'er idle. To eliminate such scenarios, we ask RabbitMQ to deliver new messages merely when the worker has acknowledged previous bulletin. The Qos function documentation can explain more.

Let's go ahead and showtime consuming messages.

You lot tin can check the arguments here. This time, nosotros will get a go channel. We can range over this channel to get messages.

We want to transport letters as JSON. To represent a chore for add operation, nosotros would define a type again in our shared.go file.

The messageChannel nosotros take, we can commencement ranging over it, decoding the message bodies into AddTask instances and and then summing Number1 and Number2 to get results.

We're launching a goroutine with the become func() call. This runs in background. So we need a way to ensure our main cli (worker) running on the foreground doesn't reach it'south terminate and quit. Nosotros tin apply a aqueduct and listen on it to continue waiting indefinitely.

Meanwhile in the goroutine, nosotros're ranging over the letters, processing the bulletin body, treatment the errors and finally acknowledging messages. While making the call to Consume, nosotros have prepare autoAck to false . So we have to manually admit a message that nosotros have processed it. If we don't acknowledge a message and worker loses connection, RabbitMQ redelivers that bulletin to other workers. This allows u.s. to gracefully retry messages even when a worker crashes.

It is also very important to recollect to manually acknowledge letters when nosotros accept auto ack off. Otherwise, RabbitMQ won't delete the messages (they are non acknowledged — and so they are non done yet), the messages will fill up upwardly RabbitMQ retention and cause mayhem. We don't desire that to happen.

This concludes our consumer. If we run become build inside our consumer directory and run ./cosumer, it should start running (given we have done everything correctly).

          ➜  consumer git:(master) ✗ ./consumer
2019/02/23 20:54:55 Consumer ready, PID: 36361

Consumer looks similar this:

Building a Publisher / Producer

Now let's create a producer that will generate random numbers and transport to the add queue.

For the publisher, we also need to repeat the connectedness, channel and queue declaration. So we can skip those parts and straight spring into the interesting parts.

You may be wondering why we're declaring the queue in both consumer and publisher — it's considering nosotros don't know which one will start commencement. So we make sure that a queue is e'er there earlier we start consuming / publishing.

We accept previously seen the AddTask type, let's generate two random numbers and create an instance. Then nosotros encode it into JSON, set for publishing to the exchange.

Let'south publish it:

The code so far looks like this:

If we build this code and run ./publisher, it will queue a task. If we have one or more consumers running, we can encounter the results.

          ➜  publisher git:(master) ✗ ./publisher
2019/02/23 21:09:59 AddTask: 221+345

And consumer window:

          ➜  consumer git:(master) ✗ ./consumer
2019/02/23 20:54:55 Consumer ready, PID: 36361
2019/02/23 21:09:59 Received a message: {"Number1":221,"Number2":345}
2019/02/23 21:09:59 Upshot of 221 + 345 is : 566
2019/02/23 21:09:59 Best-selling message

Empty Exchange Name and Queues

We publish messages to exchanges, consume from queues. In our publisher example, we didn't specify a exchange proper noun. If the exchange name is empty cord, RabbitMQ directly delivers the message to the queue passed every bit the queue name.

More Advanced Usages

In our apply case, we have used a very uncomplicated named queue. RabbitMQ exchanges can do a lot more. There are several types of message exchanges which can assist us fanout letters (deliver same message across multiple queues) or do topic based matching while delivering letters to queues. Exchanges and queues can do intelligent, configurable message routing to serve complex apply cases and build advanced distributed systems.