PHP-AMQP What's New for Friends?

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

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

How does this fit in the “Friends feed”: Let all Friends by default subscribe to some event: uploading a photo, uploading a video, blog entry, updating a profile, creating a survey, etc. When a certain event occurs, for example, uploading a photo, a message is written in your personal Exchange (Exchange), for example, the upload time, with the key “photo”.
Everyone who is Subscribed to your events receives these messages, which they read from their personal Queues. More details about the exchange broker can be found in the article “AMQP in Russian”.

Everything is much simpler if we look at a simple example:
Let the HabraBrain user have two friends: HabraLokh , HabraLuser . Users HabraLokh and HabraLuser are subscribed to all events. Let the third User HabraChaynik decided to subscribe only to update the blog.

When you enter the HabraMozin login , we create the “HabraGuru” Exchange in the broker. It is also necessary to create a queue so that our User can read the feeds 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 reset (save) the queue / exchange upon reboot.

When a Friendship relationship arises, we automatically subscribe the HabraLokh and HabraLuser users to all events of the HabraMozg User . and the User HabraChaynik went to the blog of the User HabraMozg , he liked the blog and clicked on the "Subscribe" link. In this case, the following code will be executed : Let the User HabraMozg post the presentation “How Friends' Ribbon Works” , as a result of which, at the end of the download, the following part of the code that publishes the message $ msg is executed:

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

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



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



$msg = json_encode( array('date'=> date('d-m-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 data exchange, but this is not important, you can use both csv and xml, and to which the soul lies closer

Once in 1/2/4/6/12/24 hours (whoever sets it up) it executes a kronskript , which for each user builds his personal tape. The message (title => 'FriendLenta. How to.') From the HabraBrain User will be received by the HabraLokh and HabraLuser users . And the HabraChaynik user will not receive this message. However, if the HabraBrain User decides to post on the blog a story about the " Friends Tape" , then the code will be executed: then when reading its queue ($ queueName = 'HabraTeapot'), the HabraKettle User
$queue = new AMQPQueue(APMQConection(), 'HabraLuser');

while (1){
$res = $queue->get();
if ($res['count']<0) break;
// echo "$i.{$res['msg']}";
// call QueueItemCallBack( $res['msg'] );
// обрабатываем принятые сообщения и строим "ленту Друзей"
}



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

will receive a message (title => 'How to developed the FriendLenta'), since it was sent with the key “blog” to which it was subscribed: $ queue-> bind ('HabraGuru', 'blog');

Decided to unsubscribe from messages - no problem :; The queue server works according to the FIFO principle: first come, first go. So a situation may arise when too many messages have arrived, for example, after a long vacation, then we want to count only the last thousand (out of 15 thousand). In this case: - take the queue length $ 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);
$cnn = new APMQConection();
$queue = new AMQPQueue($cnn, 'HabraTeapot');
$queue->unbind('HabraGuru', 'blog' )







As you can see, everything works reliably and simply. The “Bulletin Board” and the “News Feed” are organized on the same principle.
We subscribe to all the news in Moscow: subscribe to news about real estate in Moscow: subscribe to all the news about the real estate subscribe to real estate ads, "children's clothes" and the car: subscribe to the weather: Obviously, code examples demonstrate how easy it is to implement the "Friends , "News feed, ads or

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



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



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



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



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