PHP-AMQP What's New in Friends?

When building a social network on the type of sharding, the problem of data exchange between shards arises. Traditional replication in this case is not suitable for various reasons. The topic of sharding is a large hotel topic and is not the subject of this article.
In this architecture, it is better to use the queue server using the system: Subscription-Notification for the implementation of the “Friends tape” or “News feed”. As an exchange broker, it is proposed to use the RabbitMQ queue server, which implements the AMQP protocol, which was chosen because of its good scalability. The server part is implemented in PHP using the php-rabbit extension ( description of the API ).

Subscription-Notification architecture is an event based approach. Someone subscribes to your events. As soon as any changes occurred on your personal page: a new photo or video was uploaded, you wrote it on a blog, received a gift or just farting, i.e. an event has arisen, the Subscribers will immediately know about it through notifications.

How does it fall on the Friends tape? Let all Friends be by default subscribed to some event: photo upload, video upload, blog entry, profile update, poll creation, etc. When a certain event occurs, such as uploading a photo, a message is written to your personal Exchange (Exchange), for example, the download time, with the key “photo”.
Anyone who is Subscribed to your events receives these messages, which they read from their personal Queues. You can read more about the exchange broker in the article “AMQP in Russian”

Everything is much simpler if we consider a simple example:
Let User HabraMozg has two friends: HabraLoh , HabraLuzer . Users HabraLoh and HabraLuzer are subscribed to all events. Let the third user HabraTean decided to subscribe only to update the blog.
')
When establishing the login HabraMozg, we create a HabraGuru Exchange in the broker. It is also necessary to create a queue so that our User can read the tapes of his friends.

$cnn = new APMQConection();
$exchange = new AMQPExchange($cnn);
$exchange->declare('HabraGuru', 'topic',AMQP_DURABLE );
$queue = new AMQPQueue($cnn);
$queue ->declare('HabraGuru', AMQP_DURABLE)

The AMQP_DURABLE parameter instructs the broker not to flush (save) the queue / exchange upon reload.

When the “Friendship” relationship arises, we automatically subscribe the HabraLoh and HabraLuzer users to all events of the HabraMozg User.

$cnn = new APMQConection();
$queue = new AMQPQueue($cnn, 'HabraLoh');
$queue->bind('HabraGuru', '*' );
and
$cnn = new APMQConection();
$queue = new AMQPQueue($cnn, 'HabraLuser');
$queue->bind('HabraGuru', '*' );

And User HabraTeaGon went to the blog of User HabraMozg , he liked the blog and he clicked on the link "Subscribe". This will execute the following code:

$cnn = new APMQConection();
$queue = new AMQPQueue($cnn, 'HabraTeapot');
$queue->bind('HabraGuru', 'blog' );

Suppose User HabraMoz has laid out the presentation “How the Friends Ribbon Works” , and as a result, after the download is complete, the following part of the code is executed, which publishes the message $ msg:

$msg = json_encode( array('date'=> date('dm-y'), 'type'=> 'presentation' , title => 'FriendLenta. How to.'));
$cnn = new APMQConection();
$exchange = new AMQPExchange($cnn, 'HabraGuru');
$exchange->publish( $msg, 'presentation');
In this example, JSON is used as a data exchange, but this is not essential, you can use both csv and xml, and to which the soul is closer

Once every 1/2/4/6/12/24 hours (who will set it up), work out the kronscript, which for each user builds his personal tape.
$queue = new AMQPQueue(APMQConection(), 'HabraLuser');

while (1){
$res = $queue->get();
if ($res['count']<0) break;
// echo "$i.{$res['msg']}";
// call QueueItemCallBack( $res['msg'] );
// " "
}

The message (title => 'FriendLenta. How to.') From the User HabraMozg will be received by the users HabraLoh and HabraLuzer . And the user Habra the Teapot of the given message will not receive. However, if the User HabraMand decides to post on the blog a story about the " Friends Tape" , the code will be executed:

$msg = json_encode( array('date'=> date('dm-y'), 'type'=> 'blog' , title => 'How to developed the FriendLenta'));
$cnn = new APMQConection();
$exchange = new AMQPExchange($cnn, 'HabraGuru');
$exchange->publish( $msg, 'blog');

then when reading its queue ($ queueName = 'HabraTeapot'), the User HabraTell will receive a message (title => 'How to developed the FriendLenta'), since it was sent with the key “blog” to which it was signed: $ queue -> bind ('HabraGuru', 'blog');

We decided to unsubscribe from the messages - no problem:
$cnn = new APMQConection();
$queue = new AMQPQueue($cnn, 'HabraTeapot');
$queue->unbind('HabraGuru', 'blog' ) $cnn = new APMQConection();
$queue = new AMQPQueue($cnn, 'HabraTeapot');
$queue->unbind('HabraGuru', 'blog' ) ;

The queue server works on the FIFO principle: first come, first come out. So a situation may arise when too many messages have arrived, for example after a long vacation, we want to count only the last thousand (out of 15 thousand). In this case:
- we take the length of the queue $ len = $ queue-> declare ();
- if $ len> MAXQUEUELEN then $ dummy = $ len - MAXQUEUELEN;
- read $ dummy elements of the queue (in our case, 14 thousand);
- the remaining MAXQUEUELEN process (1 thousand messages);

As you can see, everything works reliably and simply. According to the same principle, "Bulletin Board" and "News Feed" are arranged.
We subscribe to all the news in Moscow:

$queue = new AMQPQueue(new APMQConection(), 'HabraGuru');
$queue->bind('HabraGuru', '*.msk' );

Subscribe to news about real estate in Moscow:

$queue = new AMQPQueue(new APMQConection(), 'HabraGuru');
$queue->bind('HabraGuru', 'realty.msk' );

Subscribe to all the news about real estate

$queue = new AMQPQueue(new APMQConection(), 'HabraGuru');
$queue->bind('HabraGuru', 'realty.*' );

We subscribe to real estate announcements, "children's things" and cars:

$queue = new AMQPQueue(new APMQConection(), 'HabraGuru');
$queue->bind('HabraGuru', 'realty.notes.spb' );
$queue->bind('HabraGuru', 'children.notes.spb' );
$queue->bind('HabraGuru', 'auto.notes.spb' );

We subscribe to the weather:

$queue = new AMQPQueue(new APMQConection(), 'HabraGuru');
$queue->bind('HabraGuru', 'weathe.spb' );

Obviously, code samples clearly show how easy it is to implement a “Friends feed,” a news feed, ads, or