<pre name="code" class="html">mysql master:
test:/root/zk# cat zk.pl
use ZooKeeper;
use AnyEvent;
use AE;
use Data::Dumper;
use IO::Socket;
my $zk = ZooKeeper->new(hosts => '120.55.118.6:2181, 120.55.118.6:2182, 120.55.118.6:2183');
my $stat = $zk->exists('/mysql');
unless ($stat){
$zk->create('/mysql');
}
my $created_path = $zk->create('/mysql/0001' ,
ephemeral => 1
);
print "----------------
";
print $created_path;
print "----------------
";
###获取数据
print Dumper($zk->get("/mysql/0001"));
###修改数据
$zk->set('/mysql/0001' =>'192.168.32.6');
print Dumper($zk->get("/mysql/0001"));
my $cv = AE::cv;
##所有子节点数组
my $zk_value = $zk->get('/mysql/0001', watcher => sub {
###事件状态
my $event = shift;
print "触发了事件.";
print $event->{'type'}."
";
print "事件状态.";
print $event->{'state'}."
";
$cv->send($event) }
);
print "------------------
";
print $zk_value;
print "
";
###检测3306端口
sub check_port {
( $server, $port ) = ('127.0.0.1','3306');
$sock = IO::Socket::INET->new(PeerAddr => $server,
PeerPort => $port,
Proto => 'tcp');
print "$sock
";
if ($sock)
{return 1}
else
{$zk->close; print "close zk
"; exit 0 };
};
##定义watch
my $t = AnyEvent->timer(
after => 0,
interval => 5,
cb => &check_port
);
##不要再每秒打印时间
##undef $t;
print Dumper($cv);
my $child_event = $cv->recv;
test:/root/zk#
mysql slave:
slave:/root/zk# cat zk.pl
use ZooKeeper;
use AnyEvent;
use AE;
use Data::Dumper;
use IO::Socket;
my $zk = ZooKeeper->new(hosts => '120.55.118.6:2181,120.55.118.6:2182,120.55.118.6:2183');
my $stat = $zk->exists('/mysql');
unless ($stat){
$zk->create('/mysql');
}
my $created_path = $zk->create('/mysql/0002',
ephemeral => 1
);
print "----------------
";
print $created_path;
print "----------------
";
###获取数据
print Dumper($zk->get("/mysql/0002"));
###修改数据
$zk->set('/mysql/0002' =>'192.168.32.116');
print Dumper($zk->get("/mysql/0002"));
my $cv = AE::cv;
##所有子节点数组
my $zk_value = $zk->get('/mysql/0002', watcher => sub {
###事件状态
my $event = shift;
print "触发了事件.";
print $event->{'type'}."
";
print "事件状态.";
print $event->{'state'}."
";
$cv->send($event) });
print "------------------
";
print $zk_value;
print "
";
###检测3306端口
sub check_port {
( $server, $port ) = ('127.0.0.1','3306');
$sock = IO::Socket::INET->new(PeerAddr => $server,
PeerPort => $port,
Proto => 'tcp');
print "$sock
";
if ($sock)
{return 1}
else
{$zk->close; print "close zk
"; exit 0 };
};
##定义watch
my $t = AnyEvent->timer(
after => 0,
interval => 5,
cb => &check_port
);
##不要再每秒打印时间
##undef $t;
my $child_event = $cv->recv;
测试脚本:
zjtest7-redis:/root/zk# cat test_zk.pl
use ZooKeeper;
use AnyEvent;
use AE;
use Data::Dumper;
use IO::Socket;
sub check_port {
( $server, $port ) = ('127.0.0.1','3306');
$sock = IO::Socket::INET->new(PeerAddr => $server,
PeerPort => $port,
Proto => 'tcp');
if ($sock)
{return 1}
else
{return 0 };
};
my @ip_list=();
my $zk = ZooKeeper->new(hosts => '120.55.118.6:2181,120.55.118.6:2182,120.55.118.6:2183') ;
print Dumper($zk);
eval {
my $stat = $zk->exists('/mysql/0001');
if ($stat){
$mysql_ip = $zk->get('/mysql/0001');
print $mysql_ip."
";
}
else{
$mysql_ip = $zk->get('/mysql/0002');
print $mysql_ip."
";
};
use DBI;
my $database='zjzc';
my $user="zjzc_app";
my $passwd="1234567";
my @arr2=();
my $dbh = DBI->connect("dbi:mysql:database=$database;host=$mysql_ip;port=3306",$user,$passwd,{
RaiseError => 1,
AutoCommit => 0
} ) or die "can't connect to database ". DBI-errstr;
my $hostSql = qq{select id,name from scan; };
my ($a1, $a2, $a3,$a4,$a5,$a6,$a7,$a8,$a9);
my $selStmt = $dbh->prepare($hostSql);
$selStmt->execute();
$selStmt->bind_columns(undef, $a1, $a2);
$selStmt->execute();
while( $selStmt->fetch() )
{ push (@arr2, "$a1 $a2 $a3
" );
};
print "@arr2 is @arr2
";
$dbh->disconnect;
};
第一步测试mysql 高可用；
zjtest7-redis:/root/zk# perl test_zk.pl
$VAR1 = bless( {
'default_acl' => [
{
'id' => 'anyone',
'perms' => 31,
'scheme' => 'world'
}
],
'hosts' => '120.55.118.6:2181,120.55.118.6:2182,120.55.118.6:2183',
'timeout' => 10000,
'dispatcher' => bless( {
'ae_watcher' => bless( do{(my $o = '¸¢Yʂ¿X秂9')}, 'EV::IO' ),
'watchers' => {},
'ignore_session_events' => 1,
'dispatch_cb' => sub { "DUMMY" },
'channel' => bless( {}, 'ZooKeeper::Channel' )
}, 'ZooKeeper::Dispatcher::AnyEvent' ),
'buffer_length' => 2048
}, 'ZooKeeper' );
192.168.32.6
@arr2 is 1 aaabbb
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
2 cccddeqe
此时连接的是192.168.32.6节点:
观察zk 节点信息:
[zk: 127.0.0.1(CONNECTED) 85] ls /mysql
[0002, 0001]
关闭32.6的mysql
此时32.6上的zk ：
IO::Socket::INET=GLOB(0x220a300)
IO::Socket::INET=GLOB(0x22ed950)
IO::Socket::INET=GLOB(0x203ea60)
IO::Socket::INET=GLOB(0x2131bd0)
IO::Socket::INET=GLOB(0x215fc88)
close zk
You have mail in /var/spool/mail/root
观察zk信息:
[zk: 127.0.0.1(CONNECTED) 90] ls /mysql
[0002]
zjtest7-redis:/root/zk# perl test_zk.pl
$VAR1 = bless( {
'default_acl' => [
{
'id' => 'anyone',
'scheme' => 'world',
'perms' => 31
}
],
'buffer_length' => 2048,
'hosts' => '120.55.118.6:2181,120.55.118.6:2182,120.55.118.6:2183',
'timeout' => 10000,
'dispatcher' => bless( {
'channel' => bless( {}, 'ZooKeeper::Channel' ),
'watchers' => {},
'ae_watcher' => bless( do{(my $o = 'Һ¹|%{')}, 'EV::IO' ),
'dispatch_cb' => sub { "DUMMY" },
'ignore_session_events' => 1
}, 'ZooKeeper::Dispatcher::AnyEvent' )
}, 'ZooKeeper' );
192.168.32.116
@arr2 is 99 dadsgdf243
已切换；